44 using namespace Data_Namespace;
47 namespace filesystem {
48 class directory_iterator;
52 namespace File_Namespace {
92 using TablePair = std::pair<const int32_t, const int32_t>;
105 int32_t epoch_floor{0};
106 uint32_t metadata_file_count{0};
107 uint64_t total_metadata_file_size{0};
108 uint64_t total_metadata_page_count{0};
109 std::optional<uint64_t> total_free_metadata_page_count{};
110 uint32_t data_file_count{0};
111 uint64_t total_data_file_size{0};
112 uint64_t total_data_page_count{0};
113 std::optional<uint64_t> total_free_data_page_count{};
114 std::optional<uint32_t> fragment_count{};
134 size_t source_page_num,
136 int32_t destination_file_id,
137 size_t destination_page_num)
138 : source_file_id(source_file_id)
139 , source_page_num(source_page_num)
140 , source_page_header_size(source_page_header_size)
141 , destination_file_id(destination_file_id)
142 , destination_page_num(destination_page_num) {}
160 FileMgr(
const int32_t deviceId,
163 const int32_t max_rollback_epochs = -1,
164 const size_t num_reader_threads = 0,
165 const int32_t epoch = -1,
169 FileMgr(
const int32_t deviceId,
172 const size_t defaultPageSize,
173 const bool runCoreInit);
184 const size_t numBytes = 0)
override;
186 bool isBufferOnDevice(
const ChunkKey& key)
override;
191 void deleteBuffer(
const ChunkKey& key,
const bool purge =
true)
override;
192 void deleteBuffersWithPrefix(
const ChunkKey& keyPrefix,
193 const bool purge =
true)
override;
198 void fetchBuffer(
const ChunkKey& key,
200 const size_t numBytes)
override;
210 const size_t numBytes = 0)
override;
215 virtual Page requestFreePage(
size_t pagesize,
const bool isMetadata);
219 inline std::string
printSlabs()
override {
return "Not Implemented"; }
226 return files_.at(fileId);
230 const boost::filesystem::directory_iterator& fileIterator)
const;
232 void init(
const size_t num_reader_threads,
const int32_t epochOverride);
233 void init(
const std::string& dataPathToConvertFrom,
const int32_t epochOverride);
235 void copyPage(
Page& srcPage,
238 const size_t reservedHeaderSize,
239 const size_t numBytes,
240 const size_t offset);
255 void requestFreePages(
size_t npages,
257 std::vector<Page>& pages,
258 const bool isMetadata);
261 const ChunkKey& keyPrefix)
override;
263 bool hasChunkMetadataForKeyPrefix(
const ChunkKey& keyPrefix);
270 void checkpoint()
override;
271 void checkpoint(
const int32_t db_id,
const int32_t tb_id)
override {
272 LOG(
FATAL) <<
"Operation not supported, api checkpoint() should be used instead";
280 inline virtual int32_t
epoch(int32_t db_id, int32_t tb_id)
const {
return epoch(); }
282 inline int32_t
epochFloor()
const {
return static_cast<int32_t
>(epoch_.floor()); }
285 int32_t newEpoch = epoch_.increment();
286 epochIsCheckpointed_ =
false;
290 LOG(
FATAL) <<
"Epoch for table (" << fileMgrKey_.first <<
", " << fileMgrKey_.second
291 <<
") greater than maximum allowed value of "
301 return epoch() - (epochIsCheckpointed_ ? 0 : 1);
324 FILE* getFileForFileId(
const int32_t fileId);
326 size_t getNumChunks()
override;
327 size_t getNumUsedMetadataPagesForChunkKey(
const ChunkKey& chunkKey)
const;
331 int32_t getDBVersion()
const;
332 bool getDBConvert()
const;
333 void createTopLevelMetadata();
335 virtual void closeRemovePhysical();
337 void removeTableRelatedDS(
const int32_t db_id,
const int32_t table_id)
override;
339 virtual void free_page(std::pair<FileInfo*, int32_t>&& page);
343 boost::filesystem::path getFilePath(
const std::string& file_name)
const;
346 void writePageMappingsToStatusFile(
const std::vector<PageMapping>& page_mappings);
349 void renameCompactionStatusFile(
const char*
const from_status,
350 const char*
const to_status);
357 virtual bool updatePageIfDeleted(
FileInfo* file_info,
369 virtual std::string describeSelf()
const;
371 static constexpr
size_t DEFAULT_NUM_PAGES_PER_DATA_FILE{256};
372 static constexpr
size_t DEFAULT_NUM_PAGES_PER_METADATA_FILE{4096};
375 static constexpr
char const* COPY_PAGES_STATUS{
"pending_data_compaction_0"};
376 static constexpr
char const* UPDATE_PAGE_VISIBILITY_STATUS{
"pending_data_compaction_1"};
377 static constexpr
char const* DELETE_EMPTY_FILES_STATUS{
"pending_data_compaction_2"};
381 static void setNumPagesPerDataFile(
size_t num_pages);
382 static void setNumPagesPerMetadataFile(
size_t num_pages);
384 static void renameAndSymlinkLegacyFiles(
const std::string& table_data_dir);
386 static constexpr
char LEGACY_EPOCH_FILENAME[] =
"epoch";
387 static constexpr
char EPOCH_FILENAME[] =
"epoch_metadata";
388 static constexpr
char DB_META_FILENAME[] =
"dbmeta";
389 static constexpr
char FILE_MGR_VERSION_FILENAME[] =
"filemgr_version";
390 static constexpr int32_t INVALID_VERSION = -1;
398 std::map<int32_t, FileInfo*>
406 int32_t fileMgrVersion_;
408 const int32_t latestFileMgrVersion_{2};
409 FILE* DBMetaFile_ =
nullptr;
416 bool isFullyInitted_{
false};
437 FileInfo* createFile(
const size_t pageSize,
const size_t numPages);
438 FileInfo* openExistingFile(
const std::string& path,
439 const int32_t fileId,
440 const size_t pageSize,
441 const size_t numPages,
442 std::vector<HeaderInfo>& headerVec);
443 void createEpochFile(
const std::string& epochFileName);
444 int32_t openAndReadLegacyEpochFile(
const std::string& epochFileName);
445 void openAndReadEpochFile(
const std::string& epochFileName);
446 void writeAndSyncEpochToDisk();
447 void setEpoch(
const int32_t newEpoch);
448 int32_t readVersionFromDisk(
const std::string& versionFileName)
const;
449 void writeAndSyncVersionToDisk(
const std::string& versionFileName,
451 void processFileFutures(std::vector<std::future<std::vector<HeaderInfo>>>& file_futures,
452 std::vector<HeaderInfo>& headerVec);
455 const size_t numBytes = 0);
458 const std::vector<HeaderInfo>::const_iterator& headerStartIt,
459 const std::vector<HeaderInfo>::const_iterator& headerEndIt);
462 void migrateToLatestFileMgrVersion();
463 void migrateEpochFileV0();
464 void migrateLegacyFilesV1();
468 void clearFileInfos();
471 void copySourcePageForCompaction(
const Page& source_page,
473 std::vector<PageMapping>& page_mappings,
474 std::set<Page>& touched_pages);
475 int32_t copyPageWithoutHeaderSize(
const Page& source_page,
476 const Page& destination_page);
477 void sortAndCopyFilePagesForCompaction(
size_t page_size,
478 std::vector<PageMapping>& page_mappings,
479 std::set<Page>& touched_pages);
480 void updateMappedPagesVisibility(
const std::vector<PageMapping>& page_mappings);
481 void deleteEmptyFiles();
482 void resumeFileCompaction(
const std::string& status_file_name);
483 std::vector<PageMapping> readPageMappingsFromStatusFile();
488 void closePhysicalUnlocked();
489 void syncFilesToDisk();
491 void initializeNumThreads(
size_t num_reader_threads = 0);
492 virtual FileBuffer* allocateBuffer(
const size_t page_size,
494 const size_t num_bytes = 0);
497 const std::vector<HeaderInfo>::const_iterator& headerStartIt,
498 const std::vector<HeaderInfo>::const_iterator& headerEndIt);
499 virtual ChunkKeyToChunkMap::iterator deleteBufferUnlocked(
500 const ChunkKeyToChunkMap::iterator chunk_it,
501 const bool purge =
true);
503 const size_t numBytes = 0)
const;
506 void rollOffOldData(
const int32_t epochCeiling,
const bool shouldCheckpoint);
507 void freePagesBeforeEpoch(
const int32_t min_epoch);
508 void freePagesBeforeEpochUnlocked(
const int32_t min_epoch,
518 inline int32_t
epoch()
const {
return static_cast<int32_t
>(epoch_.ceiling()); }
519 void writeDirtyBuffers();
521 void setDataAndMetadataFileStats(
StorageStats& storage_stats)
const;
522 uint32_t getFragmentCount()
const;
528 bool epochIsCheckpointed_ =
true;
529 FILE* epochFile_ =
nullptr;
DEVICE auto upper_bound(ARGS &&...args)
virtual int32_t epoch(int32_t db_id, int32_t tb_id) const
Returns current value of epoch - should be one greater than recorded at last checkpoint. Because FileMgr only contains buffers from one table we can just return the FileMgr's epoch instead of finding a table-specific epoch.
std::vector< int > ChunkKey
TablePair fileMgrKey_
Global FileMgr.
mapd_shared_mutex mutex_free_page_
virtual bool hasFileMgrKey() const
std::vector< HeaderInfo > header_infos
A logical page (Page) belongs to a file on disk.
int32_t destination_file_id
std::string printSlabs() override
size_t getMaxSize() override
std::string getFileMgrBasePath() const
std::mutex getPageMutex_
pointer to DB level metadata
Represents/provides access to contiguous data stored in the file system.
static int64_t max_allowable_epoch()
MgrType getMgrType() override
std::string fileMgrBasePath_
std::multimap< size_t, int32_t > PageSizeFileMMap
Maps logical page sizes to files.
int32_t lastCheckpointedEpoch() const
Returns value of epoch at last checkpoint.
size_t getInUseSize() override
static size_t num_pages_per_data_file_
int32_t PageHeaderSizeType
int32_t db_version_
the index of the next file id
PageMapping(int32_t source_file_id, size_t source_page_num, PageHeaderSizeType source_page_header_size, int32_t destination_file_id, size_t destination_page_num)
void init(LogOptions const &log_opts)
std::shared_timed_mutex mapd_shared_mutex
ChunkKeyToChunkMap chunkIndex_
PageSizeFileMMap fileIndex_
A map of files accessible via a file identifier.
std::vector< std::pair< FileInfo *, int32_t > > free_pages_
An AbstractBuffer is a unit of data management for a data manager.
size_t num_reader_threads_
Maps page sizes to FileInfo objects.
static size_t num_pages_per_metadata_file_
std::map< ChunkKey, FileBuffer * > ChunkKeyToChunkMap
Maps ChunkKeys (unique ids for Chunks) to Chunk objects.
#define DEFAULT_PAGE_SIZE
std::string compaction_status_file_name
const TablePair get_fileMgrKey() const
size_t defaultPageSize_
number of threads used when loading data
int32_t maxRollbackEpochs_
DEVICE auto lower_bound(ARGS &&...args)
void checkpoint(const int32_t db_id, const int32_t tb_id) override
virtual bool failOnReadError() const
True if a read error should cause a fatal error.
std::string getStringMgrType() override
size_t getAllocated() override
std::map< int32_t, FileInfo * > files_
FileInfo * getFileInfoForFileId(const int32_t fileId) const
size_t destination_page_num
int32_t maxRollbackEpochs()
Returns value max_rollback_epochs.
std::pair< const int32_t, const int32_t > TablePair
int32_t epochFloor() const
PageHeaderSizeType source_page_header_size
bool isAllocationCapped() override
mapd_shared_mutex chunkIndexMutex_
mapd_shared_mutex files_rw_mutex_
size_t getNumReaderThreads()
Returns number of threads defined by parameter num-reader-threads which should be used during initial...