#include <FileMgr.h>

Inheritance diagram for File_Namespace::FileMgr:

Collaboration diagram for File_Namespace::FileMgr:

Public Member Functions
	FileMgr (const int deviceId, GlobalFileMgr *gfm, const std::pair< const int, const int > fileMgrKey, const size_t num_reader_threads=0, const int epoch=-1, const size_t defaultPageSize=2097152)
	Constructor. More...

	FileMgr (const int deviceId, GlobalFileMgr *gfm, const std::pair< const int, const int > fileMgrKey, const bool initOnly)

	FileMgr (GlobalFileMgr *gfm, const size_t defaultPageSize, std::string basePath)

	~FileMgr () override
	Destructor. More...

AbstractBuffer *	createBuffer (const ChunkKey &key, size_t pageSize=0, const size_t numBytes=0) override
	Creates a chunk with the specified key and page size. More...

bool	isBufferOnDevice (const ChunkKey &key) override

void	deleteBuffer (const ChunkKey &key, const bool purge=true) override
	Deletes the chunk with the specified key. More...

void	deleteBuffersWithPrefix (const ChunkKey &keyPrefix, const bool purge=true) override

AbstractBuffer *	getBuffer (const ChunkKey &key, const size_t numBytes=0) override
	Returns the a pointer to the chunk with the specified key. More...

void	fetchBuffer (const ChunkKey &key, AbstractBuffer *destBuffer, const size_t numBytes) override

AbstractBuffer *	putBuffer (const ChunkKey &key, AbstractBuffer *d, const size_t numBytes=0) override
	Puts the contents of d into the Chunk with the given key. More...

AbstractBuffer *	alloc (const size_t numBytes) override

void	free (AbstractBuffer *buffer) override

Page	requestFreePage (size_t pagesize, const bool isMetadata)

MgrType	getMgrType () override

std::string	getStringMgrType () override

std::string	printSlabs () override

void	clearSlabs () override

size_t	getMaxSize () override

size_t	getInUseSize () override

size_t	getAllocated () override

bool	isAllocationCapped () override

FileInfo *	getFileInfoForFileId (const int fileId)

void	init (const size_t num_reader_threads)

void	init (const std::string dataPathToConvertFrom)

void	copyPage (Page &srcPage, FileMgr *destFileMgr, Page &destPage, const size_t reservedHeaderSize, const size_t numBytes, const size_t offset)

void	requestFreePages (size_t npages, size_t pagesize, std::vector< Page > &pages, const bool isMetadata)
	Obtains free pages – creates new files if necessary – of the requested size. More...

void	getChunkMetadataVec (std::vector< std::pair< ChunkKey, ChunkMetadata >> &chunkMetadataVec) override

void	getChunkMetadataVecForKeyPrefix (std::vector< std::pair< ChunkKey, ChunkMetadata >> &chunkMetadataVec, const ChunkKey &keyPrefix) override

void	checkpoint () override
	Fsyncs data files, writes out epoch and fsyncs that. More...

void	checkpoint (const int db_id, const int tb_id) override

int	epoch ()
	Returns current value of epoch - should be one greater than recorded at last checkpoint. More...

size_t	getNumReaderThreads ()
	Returns number of threads defined by parameter num-reader-threads which should be used during initial load and consequent read of data. More...

FILE *	getFileForFileId (const int fileId)
	Returns FILE pointer associated with requested fileId. More...

size_t	getNumChunks () override

int	getDBVersion () const
	Index for looking up chunks. More...

bool	getDBConvert () const

void	createTopLevelMetadata ()

std::string	getFileMgrBasePath () const

void	closeRemovePhysical ()

void	free_page (std::pair< FileInfo *, int > &&page)

const std::pair< const int, const int >	get_fileMgrKey () const

Public Attributes
ChunkKeyToChunkMap	chunkIndex_

Private Member Functions
FileInfo *	createFile (const size_t pageSize, const size_t numPages)
	Adds a file to the file manager repository. More...

FileInfo *	openExistingFile (const std::string &path, const int fileId, const size_t pageSize, const size_t numPages, std::vector< HeaderInfo > &headerVec)

void	createEpochFile (const std::string &epochFileName)

void	openEpochFile (const std::string &epochFileName)

void	writeAndSyncEpochToDisk ()

void	createDBMetaFile (const std::string &DBMetaFileName)

bool	openDBMetaFile (const std::string &DBMetaFileName)

void	writeAndSyncDBMetaToDisk ()

void	setEpoch (int epoch)

void	processFileFutures (std::vector< std::future< std::vector< HeaderInfo >>> &file_futures, std::vector< HeaderInfo > &headerVec)

Private Attributes
GlobalFileMgr *	gfm_

std::pair< const int, const int >	fileMgrKey_
	Global FileMgr. More...

std::string	fileMgrBasePath_

std::vector< FileInfo * >	files_

PageSizeFileMMap	fileIndex_
	A vector of files accessible via a file identifier. More...

size_t	num_reader_threads_
	Maps page sizes to FileInfo objects. More...

size_t	defaultPageSize_
	number of threads used when loading data More...

unsigned	nextFileId_

int	epoch_
	the index of the next file id More...

FILE *	epochFile_
	the current epoch (time of last checkpoint) More...

int	db_version_

FILE *	DBMetaFile_

std::mutex	getPageMutex_
	pointer to DB level metadata More...

mapd_shared_mutex	chunkIndexMutex_

mapd_shared_mutex	files_rw_mutex_

mapd_shared_mutex	mutex_free_page

std::vector< std::pair < FileInfo *, int > >	free_pages

Friends
class	GlobalFileMgr

Detailed Description

Definition at line 85 of file FileMgr.h.

Constructor & Destructor Documentation

File_Namespace::FileMgr::FileMgr	(	const int	deviceId,
		GlobalFileMgr *	gfm,
		const std::pair< const int, const int >	fileMgrKey,
		const size_t	num_reader_threads = `0`,
		const int	epoch = `-1`,
		const size_t	defaultPageSize = `2097152`
	)

Constructor.

Definition at line 67 of file FileMgr.cpp.

References init().

     : AbstractBufferMgr(deviceId)
     , gfm_(gfm)
     , fileMgrKey_(fileMgrKey)
     , defaultPageSize_(defaultPageSize)
     , nextFileId_(0)
     , epoch_(epoch) {
   init(num_reader_threads);
 }

Here is the call graph for this function:

File_Namespace::FileMgr::FileMgr	(	const int	deviceId,
		GlobalFileMgr *	gfm,
		const std::pair< const int, const int >	fileMgrKey,
		const bool	initOnly
	)

Definition at line 83 of file FileMgr.cpp.

References epochFile_, fileMgrBasePath_, fileMgrKey_, files_, File_Namespace::GlobalFileMgr::getBasePath(), gfm_, and to_string().

     : AbstractBufferMgr(deviceId)
     , gfm_(gfm)
     , fileMgrKey_(fileMgrKey)
     , defaultPageSize_(0)
     , nextFileId_(0)
     , epoch_(0) {
   const std::string fileMgrDirPrefix("table");
   const std::string FileMgrDirDelim("_");
   fileMgrBasePath_ = (gfm_->getBasePath() + fileMgrDirPrefix + FileMgrDirDelim +
                       std::to_string(fileMgrKey_.first) +                     // db_id
                       FileMgrDirDelim + std::to_string(fileMgrKey_.second));  // tb_id
   epochFile_ = nullptr;
   files_.clear();
 }

Here is the call graph for this function:

File_Namespace::FileMgr::FileMgr	(	GlobalFileMgr *	gfm,
		const size_t	defaultPageSize,
		std::string	basePath
	)

Definition at line 102 of file FileMgr.cpp.

References init().

     : AbstractBufferMgr(0)
     , gfm_(gfm)
     , fileMgrKey_(0, 0)
     , fileMgrBasePath_(basePath)
     , defaultPageSize_(defaultPageSize)
     , nextFileId_(0)
     , epoch_(-1) {
   init(basePath);
 }

Here is the call graph for this function:

File_Namespace::FileMgr::~FileMgr ( )

override

Destructor.

Definition at line 113 of file FileMgr.cpp.

References chunkIndex_, and files_.

                   {
   // checkpoint();
   // free memory used by FileInfo objects
   for (auto chunkIt = chunkIndex_.begin(); chunkIt != chunkIndex_.end(); ++chunkIt) {
     delete chunkIt->second;
   }
   for (auto file_info : files_) {
     delete file_info;
   }
 }

Member Function Documentation

AbstractBuffer * File_Namespace::FileMgr::alloc ( const size_t numBytes = 0 )

override

Definition at line 793 of file FileMgr.cpp.

References logger::FATAL, and LOG.

                                                         {
   LOG(FATAL) << "Operation not supported";
   return nullptr;  // satisfy return-type warning
 }

void File_Namespace::FileMgr::checkpoint ( )

override

Fsyncs data files, writes out epoch and fsyncs that.

Definition at line 580 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, epoch_, logger::FATAL, files_, files_rw_mutex_, free_page(), free_pages, LOG, mutex_free_page, VLOG, and writeAndSyncEpochToDisk().

                          {
   VLOG(2) << "Checkpointing epoch: " << epoch_;
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
   for (auto chunkIt = chunkIndex_.begin(); chunkIt != chunkIndex_.end(); ++chunkIt) {
     /*
     for (auto vecIt = chunkIt->first.begin(); vecIt != chunkIt->first.end(); ++vecIt) {
         std::cout << *vecIt << ",";
     }
     cout << "Is dirty: " << chunkIt->second->isDirty_ << endl;
     */
     if (chunkIt->second->is_dirty_) {
       chunkIt->second->writeMetadata(epoch_);
       chunkIt->second->clearDirtyBits();
     }
   }
   chunkIndexWriteLock.unlock();
 
   mapd_shared_lock<mapd_shared_mutex> read_lock(files_rw_mutex_);
   for (auto fileIt = files_.begin(); fileIt != files_.end(); ++fileIt) {
     int status = (*fileIt)->syncToDisk();
     if (status != 0) {
       LOG(FATAL) << "Could not sync file to disk";
     }
   }
 
   writeAndSyncEpochToDisk();
 
   mapd_unique_lock<mapd_shared_mutex> freePagesWriteLock(mutex_free_page);
   for (auto& free_page : free_pages) {
     free_page.first->freePageDeferred(free_page.second);
   }
   free_pages.clear();
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::checkpoint	(	const int	db_id,
		const int	tb_id
	)

inlineoverride

Definition at line 197 of file FileMgr.h.

References logger::FATAL, and LOG.

                                                              {
     LOG(FATAL) << "Operation not supported, api checkpoint() should be used instead";
   }

void File_Namespace::FileMgr::clearSlabs ( )

inlineoverride

Definition at line 148 of file FileMgr.h.

148 { /* noop */

149 }

void File_Namespace::FileMgr::closeRemovePhysical ( )

Definition at line 451 of file FileMgr.cpp.

References File_Namespace::close(), epochFile_, files_, getFileMgrBasePath(), and File_Namespace::renameForDelete().

Referenced by File_Namespace::GlobalFileMgr::removeTableRelatedDS().

                                   {
   for (auto file_info : files_) {
     if (file_info->f) {
       close(file_info->f);
       file_info->f = nullptr;
     }
   }
 
   if (epochFile_) {
     close(epochFile_);
     epochFile_ = nullptr;
   }
 
   /* rename for later deletion the directory containing table related data */
   File_Namespace::renameForDelete(getFileMgrBasePath());
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::copyPage	(	Page &	srcPage,
		FileMgr *	destFileMgr,
		Page &	destPage,
		const size_t	reservedHeaderSize,
		const size_t	numBytes,
		const size_t	offset
	)

Definition at line 468 of file FileMgr.cpp.

References CHECK(), defaultPageSize_, File_Namespace::Page::fileId, getFileInfoForFileId(), File_Namespace::Page::pageNum, File_Namespace::FileInfo::read(), and File_Namespace::FileInfo::write().

Referenced by init().

                                             {
   CHECK(offset + numBytes <= defaultPageSize_);
   FileInfo* srcFileInfo = getFileInfoForFileId(srcPage.fileId);
   FileInfo* destFileInfo = destFileMgr->getFileInfoForFileId(destPage.fileId);
   int8_t* buffer = new int8_t[numBytes];
 
   size_t bytesRead = srcFileInfo->read(
       srcPage.pageNum * defaultPageSize_ + offset + reservedHeaderSize, numBytes, buffer);
   CHECK(bytesRead == numBytes);
   size_t bytesWritten = destFileInfo->write(
       destPage.pageNum * defaultPageSize_ + offset + reservedHeaderSize,
       numBytes,
       buffer);
   CHECK(bytesWritten == numBytes);
   delete[] buffer;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

AbstractBuffer * File_Namespace::FileMgr::createBuffer	(	const ChunkKey &	key,
		size_t	pageSize = `0`,
		const size_t	numBytes = `0`
	)

override

Creates a chunk with the specified key and page size.

Todo:: Make all accesses to chunkIndex_ thread-safe

Definition at line 614 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, defaultPageSize_, logger::FATAL, LOG, and showChunk().

Referenced by putBuffer().

                                                              {
   size_t actualPageSize = pageSize;
   if (actualPageSize == 0) {
     actualPageSize = defaultPageSize_;
   }
   // we will do this lazily and not allocate space for the Chunk (i.e.
   // FileBuffer yet)
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
 
   if (chunkIndex_.find(key) != chunkIndex_.end()) {
     LOG(FATAL) << "Chunk already exists for key: " << showChunk(key);
   }
   chunkIndex_[key] = new FileBuffer(this, actualPageSize, key, numBytes);
   chunkIndexWriteLock.unlock();
   return (chunkIndex_[key]);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::createDBMetaFile ( const std::string & DBMetaFileName )

private

Definition at line 539 of file FileMgr.cpp.

References File_Namespace::create(), DBMetaFile_, logger::FATAL, fileMgrBasePath_, getDBVersion(), LOG, and File_Namespace::write().

Referenced by createTopLevelMetadata().

                                                               {
   std::string DBMetaFilePath(fileMgrBasePath_ + "/" + DBMetaFileName);
   if (boost::filesystem::exists(DBMetaFilePath)) {
     LOG(FATAL) << "DB metadata file `" << DBMetaFilePath << "` already exists.";
   }
   DBMetaFile_ = create(DBMetaFilePath, sizeof(int));
   int db_ver = getDBVersion();
   write(DBMetaFile_, 0, sizeof(int), (int8_t*)&db_ver);
   // LOG(INFO) << "DB metadata file has been created.";
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::createEpochFile ( const std::string & epochFileName )

private

Definition at line 490 of file FileMgr.cpp.

References File_Namespace::create(), epoch_, epochFile_, logger::FATAL, fileMgrBasePath_, LOG, and File_Namespace::write().

Referenced by init().

                                                             {
   std::string epochFilePath(fileMgrBasePath_ + "/" + epochFileName);
   if (boost::filesystem::exists(epochFilePath)) {
     LOG(FATAL) << "Epoch file `" << epochFilePath << "` already exists";
   }
   epochFile_ = create(epochFilePath, sizeof(int));
   // Write out current epoch to file - which if this
   // function is being called should be 0
   write(epochFile_, 0, sizeof(int), (int8_t*)&epoch_);
   epoch_++;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

FileInfo * File_Namespace::FileMgr::createFile	(	const size_t	pageSize,
		const size_t	numPages
	)

private

Adds a file to the file manager repository.

This method will create a FileInfo object for the file being added, and it will create the corresponding file on physical disk with the indicated number of pages pre-allocated.

A pointer to the FileInfo object is returned, which itself has a file pointer (FILE*) and a file identifier (int fileId).

Parameters

fileName	The name given to the file in physical storage.
pageSize	The logical page size for the pages in the file.
numPages	The number of logical pages to initially allocate for the file.

Returns: FileInfo* A pointer to the FileInfo object of the added file.

Definition at line 890 of file FileMgr.cpp.

References CHECK(), File_Namespace::create(), logger::FATAL, fileIndex_, fileMgrBasePath_, files_, files_rw_mutex_, LOG, and nextFileId_.

Referenced by requestFreePage(), and requestFreePages().

                                                                           {
   // check arguments
   if (pageSize == 0 || numPages == 0) {
     LOG(FATAL) << "File creation failed: pageSize and numPages must be greater than 0.";
   }
 
   // create the new file
   FILE* f = create(fileMgrBasePath_,
                    nextFileId_,
                    pageSize,
                    numPages);  // TM: not sure if I like naming scheme here - should be in
                                // separate namespace?
   CHECK(f);
 
   // instantiate a new FileInfo for the newly created file
   int fileId = nextFileId_++;
   FileInfo* fInfo =
       new FileInfo(this, fileId, f, pageSize, numPages, true);  // true means init file
   CHECK(fInfo);
 
   mapd_unique_lock<mapd_shared_mutex> write_lock(files_rw_mutex_);
   // update file manager data structures
   files_.push_back(fInfo);
   fileIndex_.insert(std::pair<size_t, int>(pageSize, fileId));
 
   CHECK(files_.back() == fInfo);  // postcondition
   return fInfo;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::createTopLevelMetadata ( )

Definition at line 983 of file FileMgr.cpp.

References createDBMetaFile(), DB_META_FILENAME, db_version_, logger::FATAL, getDBVersion(), LOG, and openDBMetaFile().

                                      {
   if (openDBMetaFile(DB_META_FILENAME)) {
     if (db_version_ > getDBVersion()) {
       LOG(FATAL) << "DB forward compatibility is not supported. Version of OmniSci "
                     "software used is older than the version of DB being read: "
                  << db_version_;
     }
   } else {
     createDBMetaFile(DB_META_FILENAME);
   }
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::deleteBuffer	(	const ChunkKey &	key,
		const bool	purge = `true`
	)

override

Deletes the chunk with the specified key.

Definition at line 639 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, logger::FATAL, LOG, and showChunk().

                                                                 {
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
   auto chunkIt = chunkIndex_.find(key);
   // ensure the Chunk exists
   if (chunkIt == chunkIndex_.end()) {
     LOG(FATAL) << "Chunk does not exist for key: " << showChunk(key);
   }
   chunkIndexWriteLock.unlock();
   // chunkIt->second->writeMetadata(-1); // writes -1 as epoch - signifies deleted
   if (purge) {
     chunkIt->second->freePages();
   }
   //@todo need a way to represent delete in non purge case
   delete chunkIt->second;
   chunkIndex_.erase(chunkIt);
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::deleteBuffersWithPrefix	(	const ChunkKey &	keyPrefix,
		const bool	purge = `true`
	)

override

Definition at line 656 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                                                                                  {
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
   auto chunkIt = chunkIndex_.lower_bound(keyPrefix);
   if (chunkIt == chunkIndex_.end()) {
     return;  // should we throw?
   }
   while (chunkIt != chunkIndex_.end() &&
          std::search(chunkIt->first.begin(),
                      chunkIt->first.begin() + keyPrefix.size(),
                      keyPrefix.begin(),
                      keyPrefix.end()) != chunkIt->first.begin() + keyPrefix.size()) {
     /*
     cout << "Freeing pages for chunk ";
     for (auto vecIt = chunkIt->first.begin(); vecIt != chunkIt->first.end(); ++vecIt) {
         std::cout << *vecIt << ",";
     }
     cout << endl;
     */
     if (purge) {
       chunkIt->second->freePages();
     }
     //@todo need a way to represent delete in non purge case
     delete chunkIt->second;
     chunkIndex_.erase(chunkIt++);
   }
 }

int File_Namespace::FileMgr::epoch ( )

inline

Returns current value of epoch - should be one greater than recorded at last checkpoint.

Definition at line 204 of file FileMgr.h.

Referenced by File_Namespace::FileBuffer::append(), File_Namespace::FileInfo::freePage(), File_Namespace::FileInfo::openExistingFile(), File_Namespace::FileBuffer::reserve(), setEpoch(), and File_Namespace::FileBuffer::write().

204 { return epoch_; }

File_Namespace::FileMgr::epoch_

int epoch_

the index of the next file id

Definition: FileMgr.h:246

Here is the caller graph for this function:

void File_Namespace::FileMgr::fetchBuffer	(	const ChunkKey &	key,
		AbstractBuffer *	destBuffer,
		const size_t	numBytes
	)

override

Definition at line 692 of file FileMgr.cpp.

                                                  {
   // reads chunk specified by ChunkKey into AbstractBuffer provided by
   // destBuffer
   if (destBuffer->isDirty()) {
     LOG(FATAL)
         << "Aborting attempt to fetch a chunk marked dirty. Chunk inconsistency for key: "
         << showChunk(key);
   }
   mapd_shared_lock<mapd_shared_mutex> chunkIndexReadLock(chunkIndexMutex_);
   auto chunkIt = chunkIndex_.find(key);
   if (chunkIt == chunkIndex_.end()) {
     LOG(FATAL) << "Chunk does not exist for key: " << showChunk(key);
   }
   chunkIndexReadLock.unlock();
 
   AbstractBuffer* chunk = chunkIt->second;
   // ChunkSize is either specified in function call with numBytes or we
   // just look at pageSize * numPages in FileBuffer
   size_t chunkSize = numBytes == 0 ? chunk->size() : numBytes;
   if (numBytes > 0 && numBytes > chunk->size()) {
     LOG(FATAL) << "Chunk retrieved for key `" << showChunk(key) << "` is smaller ("
                << chunk->size() << ") than number of bytes requested (" << numBytes
                << ")";
   }
   destBuffer->reserve(chunkSize);
   // std::cout << "After reserve chunksize: " << chunkSize << std::endl;
   if (chunk->isUpdated()) {
     chunk->read(destBuffer->getMemoryPtr(),
                 chunkSize,
                 0,
                 destBuffer->getType(),
                 destBuffer->getDeviceId());
   } else {
     chunk->read(destBuffer->getMemoryPtr() + destBuffer->size(),
                 chunkSize - destBuffer->size(),
                 destBuffer->size(),
                 destBuffer->getType(),
                 destBuffer->getDeviceId());
   }
   destBuffer->setSize(chunkSize);
   destBuffer->syncEncoder(chunk);
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::free ( AbstractBuffer * buffer )

override

Definition at line 798 of file FileMgr.cpp.

References logger::FATAL, and LOG.

                                          {
   LOG(FATAL) << "Operation not supported";
 }

void File_Namespace::FileMgr::free_page ( std::pair< FileInfo *, int > && page )

Definition at line 1000 of file FileMgr.cpp.

References free_pages, and mutex_free_page.

Referenced by checkpoint(), and File_Namespace::FileInfo::freePage().

                                                       {
   std::unique_lock<mapd_shared_mutex> lock(mutex_free_page);
   free_pages.push_back(page);
 }

Here is the caller graph for this function:

const std::pair<const int, const int> File_Namespace::FileMgr::get_fileMgrKey ( ) const

inline

Definition at line 234 of file FileMgr.h.

Referenced by File_Namespace::FileInfo::openExistingFile().

234 { return fileMgrKey_; }

File_Namespace::FileMgr::fileMgrKey_

std::pair< const int, const int > fileMgrKey_

Global FileMgr.

Definition: FileMgr.h:238

Here is the caller graph for this function:

size_t File_Namespace::FileMgr::getAllocated ( )

inlineoverride

Definition at line 152 of file FileMgr.h.

152 { return 0; }

AbstractBuffer * File_Namespace::FileMgr::getBuffer	(	const ChunkKey &	key,
		const size_t	numBytes = `0`
	)

override

Returns the a pointer to the chunk with the specified key.

Definition at line 683 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, logger::FATAL, LOG, and showChunk().

                                                                              {
   mapd_shared_lock<mapd_shared_mutex> chunkIndexReadLock(chunkIndexMutex_);
   auto chunkIt = chunkIndex_.find(key);
   if (chunkIt == chunkIndex_.end()) {
     LOG(FATAL) << "Chunk does not exist for key: " << showChunk(key);
   }
   return chunkIt->second;
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::getChunkMetadataVec ( std::vector< std::pair< ChunkKey, ChunkMetadata >> & chunkMetadataVec )

override

Definition at line 932 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                                                                    {
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
   chunkMetadataVec.reserve(chunkIndex_.size());
   for (auto chunkIt = chunkIndex_.begin(); chunkIt != chunkIndex_.end(); ++chunkIt) {
     if (chunkIt->second->has_encoder) {
       ChunkMetadata chunkMetadata;
       chunkIt->second->encoder->getMetadata(chunkMetadata);
       chunkMetadataVec.emplace_back(chunkIt->first, chunkMetadata);
     }
   }
 }

void File_Namespace::FileMgr::getChunkMetadataVecForKeyPrefix	(	std::vector< std::pair< ChunkKey, ChunkMetadata >> &	chunkMetadataVec,
		const ChunkKey &	keyPrefix
	)

override

Definition at line 945 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                                {
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(
       chunkIndexMutex_);  // is this guarding the right structure?  it look slike we oly
                           // read here for chunk
   auto chunkIt = chunkIndex_.lower_bound(keyPrefix);
   if (chunkIt == chunkIndex_.end()) {
     return;  // throw?
   }
   while (chunkIt != chunkIndex_.end() &&
          std::search(chunkIt->first.begin(),
                      chunkIt->first.begin() + keyPrefix.size(),
                      keyPrefix.begin(),
                      keyPrefix.end()) != chunkIt->first.begin() + keyPrefix.size()) {
     /*
     for (auto vecIt = chunkIt->first.begin(); vecIt != chunkIt->first.end(); ++vecIt) {
         std::cout << *vecIt << ",";
     }
     cout << endl;
     */
     if (chunkIt->second->has_encoder) {
       ChunkMetadata chunkMetadata;
       chunkIt->second->encoder->getMetadata(chunkMetadata);
       chunkMetadataVec.emplace_back(chunkIt->first, chunkMetadata);
     }
     chunkIt++;
   }
 }

bool File_Namespace::FileMgr::getDBConvert ( ) const

Definition at line 979 of file FileMgr.cpp.

References File_Namespace::GlobalFileMgr::getDBConvert(), and gfm_.

                                  {
   return gfm_->getDBConvert();
 }

Here is the call graph for this function:

int File_Namespace::FileMgr::getDBVersion ( ) const

Index for looking up chunks.

Definition at line 975 of file FileMgr.cpp.

References File_Namespace::GlobalFileMgr::getDBVersion(), and gfm_.

Referenced by createDBMetaFile(), createTopLevelMetadata(), and writeAndSyncDBMetaToDisk().

                                 {
   return gfm_->getDBVersion();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

FILE * File_Namespace::FileMgr::getFileForFileId ( const int fileId )

Returns FILE pointer associated with requested fileId.

See Also: FileBuffer

Definition at line 919 of file FileMgr.cpp.

References CHECK(), and files_.

Referenced by File_Namespace::FileBuffer::readMetadata(), and File_Namespace::FileBuffer::writeMetadata().

                                                 {
   CHECK(fileId >= 0 && static_cast<size_t>(fileId) < files_.size());
   return files_[fileId]->f;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

FileInfo* File_Namespace::FileMgr::getFileInfoForFileId ( const int fileId )

inline

Definition at line 155 of file FileMgr.h.

Referenced by File_Namespace::FileBuffer::append(), File_Namespace::FileBuffer::copyPage(), copyPage(), File_Namespace::FileBuffer::freePages(), File_Namespace::readForThread(), File_Namespace::FileBuffer::write(), and File_Namespace::FileBuffer::writeHeader().

155 { return files_[fileId]; }

File_Namespace::FileMgr::files_

std::vector< FileInfo * > files_

Definition: FileMgr.h:241

Here is the caller graph for this function:

std::string File_Namespace::FileMgr::getFileMgrBasePath ( ) const

inline

Definition at line 230 of file FileMgr.h.

Referenced by closeRemovePhysical().

230 { return fileMgrBasePath_; }

File_Namespace::FileMgr::fileMgrBasePath_

std::string fileMgrBasePath_

Definition: FileMgr.h:239

Here is the caller graph for this function:

size_t File_Namespace::FileMgr::getInUseSize ( )

inlineoverride

Definition at line 151 of file FileMgr.h.

151 { return 0; }

size_t File_Namespace::FileMgr::getMaxSize ( )

inlineoverride

Definition at line 150 of file FileMgr.h.

150 { return 0; }

MgrType File_Namespace::FileMgr::getMgrType ( )

inlineoverride

Definition at line 145 of file FileMgr.h.

145 { return FILE_MGR; };

size_t File_Namespace::FileMgr::getNumChunks ( )

inlineoverride

Definition at line 221 of file FileMgr.h.

                                         {
     // @todo should be locked - but this is more for testing now
     return chunkIndex_.size();
   }

size_t File_Namespace::FileMgr::getNumReaderThreads ( )

inline

Returns number of threads defined by parameter num-reader-threads which should be used during initial load and consequent read of data.

Definition at line 210 of file FileMgr.h.

Referenced by File_Namespace::FileBuffer::read().

210 { return num_reader_threads_; }

File_Namespace::FileMgr::num_reader_threads_

size_t num_reader_threads_

Maps page sizes to FileInfo objects.

Definition: FileMgr.h:243

Here is the caller graph for this function:

std::string File_Namespace::FileMgr::getStringMgrType ( )

inlineoverride

Definition at line 146 of file FileMgr.h.

146 { return ToString(FILE_MGR); }

void File_Namespace::FileMgr::init ( const size_t num_reader_threads )

Definition at line 124 of file FileMgr.cpp.

References CHECK_EQ, chunkIndex_, createEpochFile(), epoch_, EPOCH_FILENAME, logger::FATAL, anonymous_namespace{StringDictionary.cpp}::file_size(), fileMgrBasePath_, fileMgrKey_, File_Namespace::fileSize(), File_Namespace::GlobalFileMgr::getBasePath(), gfm_, File_Namespace::headerCompare(), logger::INFO, LOG, MAPD_FILE_EXT, nextFileId_, num_reader_threads_, openEpochFile(), openExistingFile(), processFileFutures(), timer_start(), timer_stop(), to_string(), and VLOG.

Referenced by FileMgr().

                                                   {
   // if epoch = -1 this means open from epoch file
   const std::string fileMgrDirPrefix("table");
   const std::string FileMgrDirDelim("_");
   fileMgrBasePath_ = (gfm_->getBasePath() + fileMgrDirPrefix + FileMgrDirDelim +
                       std::to_string(fileMgrKey_.first) +                     // db_id
                       FileMgrDirDelim + std::to_string(fileMgrKey_.second));  // tb_id
   boost::filesystem::path path(fileMgrBasePath_);
   if (boost::filesystem::exists(path)) {
     if (!boost::filesystem::is_directory(path)) {
       LOG(FATAL) << "Specified path '" << fileMgrBasePath_
                  << "' for table data is not a directory.";
     }
     if (epoch_ != -1) {  // if opening at previous epoch
       int epochCopy = epoch_;
       openEpochFile(EPOCH_FILENAME);
       epoch_ = epochCopy;
     } else {
       openEpochFile(EPOCH_FILENAME);
     }
 
     auto clock_begin = timer_start();
 
     boost::filesystem::directory_iterator
         endItr;  // default construction yields past-the-end
     int maxFileId = -1;
     int fileCount = 0;
     int threadCount = std::thread::hardware_concurrency();
     std::vector<HeaderInfo> headerVec;
     std::vector<std::future<std::vector<HeaderInfo>>> file_futures;
     for (boost::filesystem::directory_iterator fileIt(path); fileIt != endItr; ++fileIt) {
       if (boost::filesystem::is_regular_file(fileIt->status())) {
         // note that boost::filesystem leaves preceding dot on
         // extension - hence MAPD_FILE_EXT is ".mapd"
         std::string extension(fileIt->path().extension().string());
 
         if (extension == MAPD_FILE_EXT) {
           std::string fileStem(fileIt->path().stem().string());
           // remove trailing dot if any
           if (fileStem.size() > 0 && fileStem.back() == '.') {
             fileStem = fileStem.substr(0, fileStem.size() - 1);
           }
           size_t dotPos = fileStem.find_last_of(".");  // should only be one
           if (dotPos == std::string::npos) {
             LOG(FATAL) << "File `" << fileIt->path()
                        << "` does not carry page size information in the filename.";
           }
           int fileId = boost::lexical_cast<int>(fileStem.substr(0, dotPos));
           if (fileId > maxFileId) {
             maxFileId = fileId;
           }
           size_t pageSize =
               boost::lexical_cast<size_t>(fileStem.substr(dotPos + 1, fileStem.size()));
           std::string filePath(fileIt->path().string());
           size_t fileSize = boost::filesystem::file_size(filePath);
           CHECK_EQ(fileSize % pageSize, size_t(0));  // should be no partial pages
           size_t numPages = fileSize / pageSize;
 
           VLOG(4) << "File id: " << fileId << " Page size: " << pageSize
                   << " Num pages: " << numPages;
 
           file_futures.emplace_back(std::async(
               std::launch::async, [filePath, fileId, pageSize, numPages, this] {
                 std::vector<HeaderInfo> tempHeaderVec;
                 openExistingFile(filePath, fileId, pageSize, numPages, tempHeaderVec);
                 return tempHeaderVec;
               }));
           fileCount++;
           if (fileCount % threadCount == 0) {
             processFileFutures(file_futures, headerVec);
           }
         }
       }
     }
 
     if (file_futures.size() > 0) {
       processFileFutures(file_futures, headerVec);
     }
     int64_t queue_time_ms = timer_stop(clock_begin);
 
     LOG(INFO) << "Completed Reading table's file metadata, Elapsed time : "
               << queue_time_ms << "ms Epoch: " << epoch_ << " files read: " << fileCount
               << " table location: '" << fileMgrBasePath_ << "'";
 
     /* Sort headerVec so that all HeaderInfos
      * from a chunk will be grouped together
      * and in order of increasing PageId
      * - Version Epoch */
 
     std::sort(headerVec.begin(), headerVec.end(), headerCompare);
 
     /* Goal of next section is to find sequences in the
      * sorted headerVec of the same ChunkId, which we
      * can then initiate a FileBuffer with */
 
     VLOG(4) << "Number of Headers in Vector: " << headerVec.size();
     if (headerVec.size() > 0) {
       ChunkKey lastChunkKey = headerVec.begin()->chunkKey;
       auto startIt = headerVec.begin();
 
       for (auto headerIt = headerVec.begin() + 1; headerIt != headerVec.end();
            ++headerIt) {
         // for (auto chunkIt = headerIt->chunkKey.begin(); chunkIt !=
         // headerIt->chunkKey.end(); ++chunkIt) {
         //    std::cout << *chunkIt << " ";
         //}
 
         if (headerIt->chunkKey != lastChunkKey) {
           chunkIndex_[lastChunkKey] =
               new FileBuffer(this, /*pageSize,*/ lastChunkKey, startIt, headerIt);
           /*
           if (startIt->versionEpoch != -1) {
               cout << "not skipping bc version != -1" << endl;
               // -1 means that chunk was deleted
               // lets not read it in
               chunkIndex_[lastChunkKey] = new FileBuffer
           (this,/lastChunkKey,startIt,headerIt);
 
           }
           else {
               cout << "Skipping bc version == -1" << endl;
           }
           */
           lastChunkKey = headerIt->chunkKey;
           startIt = headerIt;
         }
       }
       // now need to insert last Chunk
       // size_t pageSize = files_[startIt->page.fileId]->pageSize;
       // cout << "Inserting last chunk" << endl;
       // if (startIt->versionEpoch != -1) {
       chunkIndex_[lastChunkKey] =
           new FileBuffer(this, /*pageSize,*/ lastChunkKey, startIt, headerVec.end());
       //}
     }
     nextFileId_ = maxFileId + 1;
     // std::cout << "next file id: " << nextFileId_ << std::endl;
   } else {
     if (!boost::filesystem::create_directory(path)) {
       LOG(FATAL) << "Could not create data directory: " << path;
     }
     createEpochFile(EPOCH_FILENAME);
   }
 
   /* define number of reader threads to be used */
   size_t num_hardware_based_threads =
       std::thread::hardware_concurrency();  // # of threads is based on # of cores on the
                                             // host
   if (num_reader_threads == 0) {            // # of threads has not been defined by user
     num_reader_threads_ = num_hardware_based_threads;
   } else {
     if (num_reader_threads > num_hardware_based_threads) {
       num_reader_threads_ = num_hardware_based_threads;
     } else {
       num_reader_threads_ = num_reader_threads;
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::init ( const std::string dataPathToConvertFrom )

Definition at line 297 of file FileMgr.cpp.

References CHECK(), chunkIndex_, copyPage(), epoch_, EPOCH_FILENAME, File_Namespace::MultiPage::epochs, logger::FATAL, anonymous_namespace{StringDictionary.cpp}::file_size(), File_Namespace::fileSize(), File_Namespace::GlobalFileMgr::getFileMgr(), File_Namespace::FileBuffer::getMultiPage(), gfm_, File_Namespace::headerCompare(), LOG, MAPD_FILE_EXT, File_Namespace::FileBuffer::multiPages_, nextFileId_, openEpochFile(), openExistingFile(), File_Namespace::FileBuffer::pageDataSize(), File_Namespace::FileBuffer::pageSize(), processFileFutures(), requestFreePage(), File_Namespace::FileBuffer::reservedHeaderSize(), Data_Namespace::AbstractBuffer::setDirty(), Data_Namespace::AbstractBuffer::setSize(), File_Namespace::FileBuffer::size(), Data_Namespace::AbstractBuffer::syncEncoder(), and File_Namespace::FileBuffer::writeHeader().

                                                         {
   int converted_data_epoch = 0;
   boost::filesystem::path path(dataPathToConvertFrom);
   if (boost::filesystem::exists(path)) {
     if (!boost::filesystem::is_directory(path)) {
       LOG(FATAL) << "Specified path `" << path << "` is not a directory.";
     }
 
     if (epoch_ != -1) {  // if opening at previous epoch
       int epochCopy = epoch_;
       openEpochFile(EPOCH_FILENAME);
       epoch_ = epochCopy;
     } else {
       openEpochFile(EPOCH_FILENAME);
     }
 
     boost::filesystem::directory_iterator
         endItr;  // default construction yields past-the-end
     int maxFileId = -1;
     int fileCount = 0;
     int threadCount = std::thread::hardware_concurrency();
     std::vector<HeaderInfo> headerVec;
     std::vector<std::future<std::vector<HeaderInfo>>> file_futures;
     for (boost::filesystem::directory_iterator fileIt(path); fileIt != endItr; ++fileIt) {
       if (boost::filesystem::is_regular_file(fileIt->status())) {
         // note that boost::filesystem leaves preceding dot on
         // extension - hence MAPD_FILE_EXT is ".mapd"
         std::string extension(fileIt->path().extension().string());
 
         if (extension == MAPD_FILE_EXT) {
           std::string fileStem(fileIt->path().stem().string());
           // remove trailing dot if any
           if (fileStem.size() > 0 && fileStem.back() == '.') {
             fileStem = fileStem.substr(0, fileStem.size() - 1);
           }
           size_t dotPos = fileStem.find_last_of(".");  // should only be one
           if (dotPos == std::string::npos) {
             LOG(FATAL) << "File `" + fileIt->path().string() +
                               "` does not carry page size information in the filename.";
           }
           int fileId = boost::lexical_cast<int>(fileStem.substr(0, dotPos));
           if (fileId > maxFileId) {
             maxFileId = fileId;
           }
           size_t pageSize =
               boost::lexical_cast<size_t>(fileStem.substr(dotPos + 1, fileStem.size()));
           std::string filePath(fileIt->path().string());
           size_t fileSize = boost::filesystem::file_size(filePath);
           CHECK(fileSize % pageSize == 0);  // should be no partial pages
           size_t numPages = fileSize / pageSize;
 
           file_futures.emplace_back(std::async(
               std::launch::async, [filePath, fileId, pageSize, numPages, this] {
                 std::vector<HeaderInfo> tempHeaderVec;
                 openExistingFile(filePath, fileId, pageSize, numPages, tempHeaderVec);
                 return tempHeaderVec;
               }));
           fileCount++;
           if (fileCount % threadCount) {
             processFileFutures(file_futures, headerVec);
           }
         }
       }
     }
 
     if (file_futures.size() > 0) {
       processFileFutures(file_futures, headerVec);
     }
 
     /* Sort headerVec so that all HeaderInfos
      * from a chunk will be grouped together
      * and in order of increasing PageId
      * - Version Epoch */
 
     std::sort(headerVec.begin(), headerVec.end(), headerCompare);
 
     /* Goal of next section is to find sequences in the
      * sorted headerVec of the same ChunkId, which we
      * can then initiate a FileBuffer with */
 
     if (headerVec.size() > 0) {
       ChunkKey lastChunkKey = headerVec.begin()->chunkKey;
       auto startIt = headerVec.begin();
 
       for (auto headerIt = headerVec.begin() + 1; headerIt != headerVec.end();
            ++headerIt) {
         if (headerIt->chunkKey != lastChunkKey) {
           FileMgr* c_fm_ =
               dynamic_cast<File_Namespace::FileMgr*>(gfm_->getFileMgr(lastChunkKey));
           CHECK(c_fm_);
           FileBuffer* srcBuf = new FileBuffer(this, lastChunkKey, startIt, headerIt);
           chunkIndex_[lastChunkKey] = srcBuf;
           FileBuffer* destBuf = new FileBuffer(c_fm_, srcBuf->pageSize(), lastChunkKey);
           c_fm_->chunkIndex_[lastChunkKey] = destBuf;
           destBuf->syncEncoder(srcBuf);
           destBuf->setSize(srcBuf->size());
           destBuf->setDirty();  // this needs to be set to force writing out metadata
                                 // files from "checkpoint()" call
 
           size_t totalNumPages = srcBuf->getMultiPage().size();
           for (size_t pageNum = 0; pageNum < totalNumPages; pageNum++) {
             Page srcPage = srcBuf->getMultiPage()[pageNum].current();
             Page destPage = c_fm_->requestFreePage(
                 srcBuf->pageSize(),
                 false);  // may modify and use api "FileBuffer::addNewMultiPage" instead
             MultiPage multiPage(srcBuf->pageSize());
             multiPage.epochs.push_back(converted_data_epoch);
             multiPage.pageVersions.push_back(destPage);
             destBuf->multiPages_.push_back(multiPage);
             size_t reservedHeaderSize = srcBuf->reservedHeaderSize();
             copyPage(
                 srcPage, c_fm_, destPage, reservedHeaderSize, srcBuf->pageDataSize(), 0);
             destBuf->writeHeader(destPage, pageNum, converted_data_epoch, false);
           }
           lastChunkKey = headerIt->chunkKey;
           startIt = headerIt;
         }
       }
 
       // now need to insert last Chunk
       FileMgr* c_fm_ =
           dynamic_cast<File_Namespace::FileMgr*>(gfm_->getFileMgr(lastChunkKey));
       FileBuffer* srcBuf = new FileBuffer(this, lastChunkKey, startIt, headerVec.end());
       chunkIndex_[lastChunkKey] = srcBuf;
       FileBuffer* destBuf = new FileBuffer(c_fm_, srcBuf->pageSize(), lastChunkKey);
       c_fm_->chunkIndex_[lastChunkKey] = destBuf;
       destBuf->syncEncoder(srcBuf);
       destBuf->setSize(srcBuf->size());
       destBuf->setDirty();  // this needs to be set to write out metadata file from the
                             // "checkpoint()" call
 
       size_t totalNumPages = srcBuf->getMultiPage().size();
       for (size_t pageNum = 0; pageNum < totalNumPages; pageNum++) {
         Page srcPage = srcBuf->getMultiPage()[pageNum].current();
         Page destPage = c_fm_->requestFreePage(
             srcBuf->pageSize(),
             false);  // may modify and use api "FileBuffer::addNewMultiPage" instead
         MultiPage multiPage(srcBuf->pageSize());
         multiPage.epochs.push_back(converted_data_epoch);
         multiPage.pageVersions.push_back(destPage);
         destBuf->multiPages_.push_back(multiPage);
         size_t reservedHeaderSize = srcBuf->reservedHeaderSize();
         copyPage(srcPage, c_fm_, destPage, reservedHeaderSize, srcBuf->pageDataSize(), 0);
         destBuf->writeHeader(destPage, pageNum, converted_data_epoch, false);
       }
     }
     nextFileId_ = maxFileId + 1;
   } else {
     if (!boost::filesystem::create_directory(path)) {
       LOG(FATAL) << "Specified path does not exist: " << path;
     }
   }
 }

Here is the call graph for this function:

bool File_Namespace::FileMgr::isAllocationCapped ( )

inlineoverride

Definition at line 153 of file FileMgr.h.

153 { return false; }

bool File_Namespace::FileMgr::isBufferOnDevice ( const ChunkKey & key )

override

Definition at line 634 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                                                   {
   mapd_shared_lock<mapd_shared_mutex> chunkIndexReadLock(chunkIndexMutex_);
   return chunkIndex_.find(key) != chunkIndex_.end();
 }

bool File_Namespace::FileMgr::openDBMetaFile ( const std::string & DBMetaFileName )

private

Definition at line 550 of file FileMgr.cpp.

References db_version_, DBMetaFile_, anonymous_namespace{StringDictionary.cpp}::file_size(), fileMgrBasePath_, File_Namespace::open(), and File_Namespace::read().

Referenced by createTopLevelMetadata().

                                                             {
   std::string DBMetaFilePath(fileMgrBasePath_ + "/" + DBMetaFileName);
 
   if (!boost::filesystem::exists(DBMetaFilePath)) {
     // LOG(INFO) << "DB metadata file does not exist, one will be created.";
     return false;
   }
   if (!boost::filesystem::is_regular_file(DBMetaFilePath)) {
     // LOG(INFO) << "DB metadata file is not a regular file, one will be created.";
     return false;
   }
   if (boost::filesystem::file_size(DBMetaFilePath) < 4) {
     // LOG(INFO) << "DB metadata file is not sized properly, one will be created.";
     return false;
   }
   DBMetaFile_ = open(DBMetaFilePath);
   read(DBMetaFile_, 0, sizeof(int), (int8_t*)&db_version_);
 
   return true;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::openEpochFile ( const std::string & epochFileName )

private

Definition at line 502 of file FileMgr.cpp.

References epoch_, epochFile_, logger::FATAL, anonymous_namespace{StringDictionary.cpp}::file_size(), fileMgrBasePath_, LOG, File_Namespace::open(), and File_Namespace::read().

Referenced by init().

                                                           {
   std::string epochFilePath(fileMgrBasePath_ + "/" + epochFileName);
   if (!boost::filesystem::exists(epochFilePath)) {
     LOG(FATAL) << "Epoch file `" << epochFilePath << "` does not exist";
   }
   if (!boost::filesystem::is_regular_file(epochFilePath)) {
     LOG(FATAL) << "Epoch file `" << epochFilePath << "` is not a regular file";
   }
   if (boost::filesystem::file_size(epochFilePath) < 4) {
     LOG(FATAL) << "Epoch file `" << epochFilePath
                << "` is not sized properly (current size: "
                << boost::filesystem::file_size(epochFilePath) << ", expected size: 4)";
   }
   epochFile_ = open(epochFilePath);
   read(epochFile_, 0, sizeof(int), (int8_t*)&epoch_);
   // std::cout << "Epoch after open file: " << epoch_ << std::endl;
   epoch_++;  // we are in new epoch from last checkpoint
 }

Here is the call graph for this function:

Here is the caller graph for this function:

FileInfo * File_Namespace::FileMgr::openExistingFile	(	const std::string &	path,
		const int	fileId,
		const size_t	pageSize,
		const size_t	numPages,
		std::vector< HeaderInfo > &	headerVec
	)

private

Definition at line 871 of file FileMgr.cpp.

References epoch_, fileIndex_, files_, files_rw_mutex_, File_Namespace::open(), and File_Namespace::FileInfo::openExistingFile().

Referenced by init().

                                                                       {
   FILE* f = open(path);
   FileInfo* fInfo = new FileInfo(
       this, fileId, f, pageSize, numPages, false);  // false means don't init file
 
   fInfo->openExistingFile(headerVec, epoch_);
   mapd_unique_lock<mapd_shared_mutex> write_lock(files_rw_mutex_);
   if (fileId >= static_cast<int>(files_.size())) {
     files_.resize(fileId + 1);
   }
   files_[fileId] = fInfo;
   fileIndex_.insert(std::pair<size_t, int>(pageSize, fileId));
   return fInfo;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::string File_Namespace::FileMgr::printSlabs ( )

inlineoverride

Definition at line 147 of file FileMgr.h.

147 { return "Not Implemented"; }

void File_Namespace::FileMgr::processFileFutures	(	std::vector< std::future< std::vector< HeaderInfo >>> &	file_futures,
		std::vector< HeaderInfo > &	headerVec
	)

private

Definition at line 283 of file FileMgr.cpp.

Referenced by init().

                                       {
   for (auto& file_future : file_futures) {
     file_future.wait();
   }
   // concatenate the vectors after thread completes
   for (auto& file_future : file_futures) {
     auto tempHeaderVec = file_future.get();
     headerVec.insert(headerVec.end(), tempHeaderVec.begin(), tempHeaderVec.end());
   }
   file_futures.clear();
 }

Here is the caller graph for this function:

AbstractBuffer * File_Namespace::FileMgr::putBuffer	(	const ChunkKey &	key,
		AbstractBuffer *	d,
		const size_t	numBytes = `0`
	)

override

Puts the contents of d into the Chunk with the given key.

Parameters

key	- Unique identifier for a Chunk.
d	- An object representing the source data for the Chunk.

Returns: AbstractBuffer*

Definition at line 737 of file FileMgr.cpp.

                                                           {
   // obtain a pointer to the Chunk
   mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
   auto chunkIt = chunkIndex_.find(key);
   AbstractBuffer* chunk;
   if (chunkIt == chunkIndex_.end()) {
     chunk = createBuffer(key, defaultPageSize_);
   } else {
     chunk = chunkIt->second;
   }
   chunkIndexWriteLock.unlock();
   size_t oldChunkSize = chunk->size();
   // write the buffer's data to the Chunk
   // size_t newChunkSize = numBytes == 0 ? srcBuffer->size() : numBytes;
   size_t newChunkSize = numBytes == 0 ? srcBuffer->size() : numBytes;
   if (chunk->isDirty()) {
     // multiple appends are allowed,
     // but only single update is allowed
     if (srcBuffer->isUpdated() && chunk->isUpdated()) {
       LOG(FATAL) << "Aborting attempt to write a chunk marked dirty. Chunk inconsistency "
                     "for key: "
                  << showChunk(key);
     }
   }
   if (srcBuffer->isUpdated()) {
     // chunk size is not changed when fixed rows are updated or are marked as deleted.
     // but when rows are vacuumed or varlen rows are updated (new rows are appended),
     // chunk size will change. For vacuum, checkpoint should sync size from cpu to disk.
     // For varlen update, it takes another route via fragmenter using disk-level buffer.
     if (0 == numBytes && !chunk->isDirty()) {
       chunk->setSize(newChunkSize);
     }
     //@todo use dirty flags to only flush pages of chunk that need to
     // be flushed
     chunk->write((int8_t*)srcBuffer->getMemoryPtr(),
                  newChunkSize,
                  0,
                  srcBuffer->getType(),
                  srcBuffer->getDeviceId());
   } else if (srcBuffer->isAppended()) {
     CHECK_LT(oldChunkSize, newChunkSize);
     chunk->append((int8_t*)srcBuffer->getMemoryPtr() + oldChunkSize,
                   newChunkSize - oldChunkSize,
                   srcBuffer->getType(),
                   srcBuffer->getDeviceId());
   }
   // chunk->clearDirtyBits(); // Hack: because write and append will set dirty bits
   //@todo commenting out line above will make sure this metadata is set
   // but will trigger error on fetch chunk
   srcBuffer->clearDirtyBits();
   chunk->syncEncoder(srcBuffer);
   return chunk;
 }

Here is the call graph for this function:

Page File_Namespace::FileMgr::requestFreePage	(	size_t	pagesize,
		const bool	isMetadata
	)

Definition at line 802 of file FileMgr.cpp.

References CHECK(), createFile(), File_Namespace::FileInfo::fileId, fileIndex_, files_, File_Namespace::FileInfo::getFreePage(), getPageMutex_, MAX_FILE_N_METADATA_PAGES, and MAX_FILE_N_PAGES.

Referenced by File_Namespace::FileBuffer::addNewMultiPage(), init(), File_Namespace::FileBuffer::write(), and File_Namespace::FileBuffer::writeMetadata().

                                                                     {
   std::lock_guard<std::mutex> lock(getPageMutex_);
 
   auto candidateFiles = fileIndex_.equal_range(pageSize);
   int pageNum = -1;
   for (auto fileIt = candidateFiles.first; fileIt != candidateFiles.second; ++fileIt) {
     FileInfo* fileInfo = files_[fileIt->second];
     pageNum = fileInfo->getFreePage();
     if (pageNum != -1) {
       return (Page(fileInfo->fileId, pageNum));
     }
   }
   // if here then we need to add a file
   FileInfo* fileInfo;
   if (isMetadata) {
     fileInfo = createFile(pageSize, MAX_FILE_N_METADATA_PAGES);
   } else {
     fileInfo = createFile(pageSize, MAX_FILE_N_PAGES);
   }
   pageNum = fileInfo->getFreePage();
   CHECK(pageNum != -1);
   return (Page(fileInfo->fileId, pageNum));
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileMgr::requestFreePages	(	size_t	npages,
		size_t	pagesize,
		std::vector< Page > &	pages,
		const bool	isMetadata
	)

Obtains free pages – creates new files if necessary – of the requested size.

Given a page size and number of pages, this method updates the vector "pages" to include free pages of the requested size. These pages are immediately removed from the free list of the affected file(s). If there are not enough pages available among current files, new files are created and their pages are included in the vector.

Parameters

npages	The number of free pages requested
pagesize	The size of each requested page
pages	A vector containing the free pages obtained by this method

Definition at line 826 of file FileMgr.cpp.

References CHECK(), createFile(), File_Namespace::FileInfo::fileId, fileIndex_, files_, File_Namespace::FileInfo::getFreePage(), getPageMutex_, MAX_FILE_N_METADATA_PAGES, and MAX_FILE_N_PAGES.

                                                       {
   // not used currently
   // @todo add method to FileInfo to get more than one page
   std::lock_guard<std::mutex> lock(getPageMutex_);
   auto candidateFiles = fileIndex_.equal_range(pageSize);
   size_t numPagesNeeded = numPagesRequested;
   for (auto fileIt = candidateFiles.first; fileIt != candidateFiles.second; ++fileIt) {
     FileInfo* fileInfo = files_[fileIt->second];
     int pageNum;
     do {
       pageNum = fileInfo->getFreePage();
       if (pageNum != -1) {
         pages.emplace_back(fileInfo->fileId, pageNum);
         numPagesNeeded--;
       }
     } while (pageNum != -1 && numPagesNeeded > 0);
     if (numPagesNeeded == 0) {
       break;
     }
   }
   while (numPagesNeeded > 0) {
     FileInfo* fileInfo;
     if (isMetadata) {
       fileInfo = createFile(pageSize, MAX_FILE_N_METADATA_PAGES);
     } else {
       fileInfo = createFile(pageSize, MAX_FILE_N_PAGES);
     }
     int pageNum;
     do {
       pageNum = fileInfo->getFreePage();
       if (pageNum != -1) {
         pages.emplace_back(fileInfo->fileId, pageNum);
         numPagesNeeded--;
       }
     } while (pageNum != -1 && numPagesNeeded > 0);
     if (numPagesNeeded == 0) {
       break;
     }
   }
   CHECK(pages.size() == numPagesRequested);
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::setEpoch ( int epoch )

private

Definition at line 995 of file FileMgr.cpp.

References epoch(), epoch_, and writeAndSyncEpochToDisk().

                                 {
   epoch_ = epoch;
   writeAndSyncEpochToDisk();
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::writeAndSyncDBMetaToDisk ( )

private

Definition at line 571 of file FileMgr.cpp.

References DBMetaFile_, logger::FATAL, getDBVersion(), LOG, and File_Namespace::write().

                                        {
   int db_ver = getDBVersion();
   write(DBMetaFile_, 0, sizeof(int), (int8_t*)&db_ver);
   int status = fflush(DBMetaFile_);
   if (status != 0) {
     LOG(FATAL) << "Could not sync DB metadata file to disk";
   }
 }

Here is the call graph for this function:

void File_Namespace::FileMgr::writeAndSyncEpochToDisk ( )

private

Definition at line 521 of file FileMgr.cpp.

References epoch_, epochFile_, logger::FATAL, LOG, and File_Namespace::write().

Referenced by checkpoint(), and setEpoch().

                                       {
   write(epochFile_, 0, sizeof(int), (int8_t*)&epoch_);
   int status = fflush(epochFile_);
   // int status = fcntl(fileno(epochFile_),51);
   if (status != 0) {
     LOG(FATAL) << "Could not flush epoch file to disk";
   }
 #ifdef __APPLE__
   status = fcntl(fileno(epochFile_), 51);
 #else
   status = fsync(fileno(epochFile_));
 #endif
   if (status != 0) {
     LOG(FATAL) << "Could not sync epoch file to disk";
   }
   ++epoch_;
 }