RecordFileReader.h

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#pragma once
 
#include <array>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <thread>
#include <vector>
 
#include "FileFormat.h"
#include "FileHandler.h"
#include "IndexRecord.h"
#include "Record.h"
#include "RecordFormat.h"
#include "RecordReaders.h"
#include "Recordable.h"
#include "StreamId.h"
 
namespace vrs {
 
using std::map;
using std::set;
using std::string;
using std::vector;
 
class DataLayout;
class StreamPlayer;
class ProgressLogger;
 
constexpr const char* kFailFastOnIncompleteIndex = "fail_fast_on_incomplete_index";
 
class RecordFileReader {
 public:
  RecordFileReader();
  RecordFileReader(const RecordFileReader&) = delete;
  RecordFileReader(RecordFileReader&&) = delete;
 
  virtual ~RecordFileReader();
 
  RecordFileReader& operator=(const RecordFileReader&) = delete;
  RecordFileReader& operator=(RecordFileReader&&) = delete;
 
  bool isVrsFile(const string& filePath);
  bool isVrsFile(const FileSpec& fileSpec);
 
  int openFile(const string& filePath, bool autoWriteFixedIndex = false);
  int openFile(const FileSpec& fileSpec, bool autoWriteFixedIndex = false);
 
  bool isOpened() const;
 
  void setStreamPlayer(StreamId streamId, StreamPlayer* streamPlayer);
 
  int clearStreamPlayers();
 
  bool prefetchRecordSequence(
      const vector<const IndexRecord::RecordInfo*>& records,
      bool clearSequence = true);
 
  bool isRecordAvailableOrPrefetch(const IndexRecord::RecordInfo& recordInfo);
 
  int readRecord(const IndexRecord::RecordInfo& recordInfo);
 
  int readRecord(
      const IndexRecord::RecordInfo& recordInfo,
      StreamPlayer* streamPlayer,
      bool setupPlayer = false);
 
  int readAllRecords();
 
  vector<std::pair<string, int64_t>> getFileChunks() const;
 
  int64_t getTotalSourceSize() const;
 
  int closeFile();
 
  bool hasIndex() const;
 
  const set<StreamId>& getStreams() const {
    return streamIds_;
  }
 
  vector<StreamId> getStreams(RecordableTypeId typeId, const string& flavor = {}) const;
 
  StreamId getStreamForType(RecordableTypeId typeId, uint32_t indexNumber = 0) const;
 
  StreamId getStreamForName(const string& name) const;
 
  StreamId
  getStreamForFlavor(RecordableTypeId typeId, const string& flavor, uint32_t indexNumber = 0) const;
 
  StreamId getStreamForTag(
      const string& tagName,
      const string& tag,
      RecordableTypeId typeId = RecordableTypeId::Undefined) const;
 
  StreamId getStreamForSerialNumber(const string& streamSerialNumber) const;
 
  const vector<IndexRecord::RecordInfo>& getIndex() const {
    return recordIndex_;
  }
 
  const vector<const IndexRecord::RecordInfo*>& getIndex(StreamId streamId) const;
 
  uint32_t getRecordCount() const {
    return static_cast<uint32_t>(recordIndex_.size());
  }
 
  uint32_t getRecordCount(StreamId streamId) const;
 
  uint32_t getRecordCount(StreamId streamId, Record::Type recordType) const;
 
  const IndexRecord::RecordInfo* getRecord(uint32_t globalIndex) const;
 
  const IndexRecord::RecordInfo* getRecord(StreamId streamId, uint32_t indexNumber) const;
 
  const IndexRecord::RecordInfo*
  getRecord(StreamId streamId, Record::Type recordType, uint32_t indexNumber) const;
 
  const IndexRecord::RecordInfo* getLastRecord(StreamId streamId, Record::Type recordType) const;
 
  const IndexRecord::RecordInfo* getRecordByTime(double timestamp) const;
  const IndexRecord::RecordInfo* getRecordByTime(Record::Type recordType, double timestamp) const;
  const IndexRecord::RecordInfo* getRecordByTime(StreamId streamId, double timestamp) const;
  const IndexRecord::RecordInfo*
  getRecordByTime(StreamId streamId, Record::Type recordType, double timestamp) const;
 
  const IndexRecord::RecordInfo* getNearestRecordByTime(
      double timestamp,
      double epsilon,
      StreamId streamId = {},
      Record::Type recordType = Record::Type::UNDEFINED) const;
 
  uint32_t getRecordIndex(const IndexRecord::RecordInfo* record) const;
 
  uint32_t getRecordStreamIndex(const IndexRecord::RecordInfo* record) const;
 
  uint32_t getRecordSize(uint32_t recordIndex, bool useBoundaries = true) const;
 
  double getFirstDataRecordTime() const;
 
  double getLastDataRecordTime() const;
 
  bool readFirstConfigurationRecord(StreamId streamId, StreamPlayer* streamPlayer = nullptr);
 
  bool readFirstConfigurationRecords(StreamPlayer* streamPlayer = nullptr);
 
  bool readFirstConfigurationRecordsForType(
      RecordableTypeId typeId,
      StreamPlayer* streamPlayer = nullptr);
 
  const map<string, string>& getTags() const {
    return fileTags_;
  }
 
  const string& getTag(const string& name) const {
    return getTag(fileTags_, name);
  }
 
  const StreamTags& getTags(StreamId streamId) const;
 
  const map<StreamId, StreamTags>& getStreamTags() const {
    return streamTags_;
  }
 
  const string& getTag(StreamId streamId, const string& name) const {
    return getTag(getTags(streamId).user, name);
  }
 
  const string& getOriginalRecordableTypeName(StreamId streamId) const;
 
  const string& getFlavor(StreamId streamId) const;
 
  const string& getSerialNumber(StreamId streamId) const;
 
  string getStreamsSignature() const;
 
  bool mightContainImages(StreamId streamId) const;
 
  bool mightContainAudio(StreamId streamId) const;
 
  bool getRecordFormat(
      StreamId streamId,
      Record::Type recordType,
      uint32_t formatVersion,
      RecordFormat& outFormat) const;
 
  uint32_t getRecordFormats(StreamId streamId, RecordFormatMap& outFormats) const;
  std::unique_ptr<DataLayout> getDataLayout(StreamId streamId, const ContentBlockId& blockId) const;
 
  void setOpenProgressLogger(ProgressLogger* progressLogger);
 
  bool setCachingStrategy(CachingStrategy cachingStrategy) {
    return file_->setCachingStrategy(cachingStrategy);
  }
  CachingStrategy getCachingStrategy() const {
    return file_->getCachingStrategy();
  }
 
  bool setStatsCallback(const FileHandler::CacheStatsCallbackFunction& statsCallback) {
    return file_->setStatsCallback(statsCallback);
  }
 
  bool purgeFileCache() {
    return file_->purgeCache();
  }
 
 
  void setFileHandler(std::unique_ptr<FileHandler> fileHandler);
 
  std::unique_ptr<FileHandler> getFileHandler() const;
 
  static int
  vrsFilePathToFileSpec(const string& filePath, FileSpec& outFileSpec, bool checkLocalFile = false);
 
  class RecordTypeCounter : public std::array<uint32_t, enumCount<Record::Type>()> {
    using ParentType = std::array<uint32_t, enumCount<Record::Type>()>;
 
   public:
    RecordTypeCounter() : array() {
      fill(0);
    }
    inline uint32_t operator[](Record::Type recordType) const {
      return ParentType::operator[](static_cast<uint32_t>(recordType));
    }
    inline uint32_t& operator[](Record::Type recordType) {
      return ParentType::operator[](static_cast<uint32_t>(recordType));
    }
    uint32_t totalCount() const;
  };
 
  void buildRecordBoundaries(bool boundariesAndLimits = false) const; 
 
 private:
  int doOpenFile(const FileSpec& fileSpec, bool autoWriteFixedIndex, bool checkSignatureOnly);
  int readFileHeader(const FileSpec& fileSpec, FileFormat::FileHeader& outFileHeader);
  int readFileDetails(
      const FileSpec& fileSpec,
      bool autoWriteFixedIndex,
      FileFormat::FileHeader& fileHeader);
  bool readConfigRecords(
      const set<const IndexRecord::RecordInfo*>& configRecords,
      StreamPlayer* streamPlayer);
 
  static const string& getTag(const map<string, string>& tags, const string& name); 
  bool mightContainContentTypeInDataRecord(StreamId streamId, ContentType type) const; 
 
  int64_t getFollowingRecordOffset(uint32_t recordIndex, bool useBoundaries) const; 
  mutable vector<int64_t> recordBoundaries_; 
  mutable map<uint32_t, int64_t> recordLimits_; 
 
  // Members to read an open VRS file
  std::unique_ptr<FileHandler> file_;
  UncompressedRecordReader uncompressedRecordReader_;
  CompressedRecordReader compressedRecordReader_;
 
  // Source of truth describing the VRS file: must never change while the file is open.
  set<StreamId> streamIds_;
  map<StreamId, StreamTags> streamTags_;
  map<string, string> fileTags_;
  vector<IndexRecord::RecordInfo> recordIndex_;
  mutable map<StreamId, RecordTypeCounter> streamRecordCounts_;
 
  // Pointers to stream players to notify when reading records. These are NOT owned by the class.
  map<StreamId, StreamPlayer*> streamPlayers_;
 
  // Misc less critical members, for presentation or optimization
  unique_ptr<ProgressLogger> defaultProgressLogger_;
  ProgressLogger* openProgressLogger_;
  unique_ptr<std::thread> detailsSaveThread_;
  mutable map<StreamId, vector<const IndexRecord::RecordInfo*>> streamIndex_;
  // Location of the last record searched for a specific stream & record type
  // The pair: index of the record for the type (query), index of the record in the stream (result)
  mutable map<pair<StreamId, Record::Type>, pair<uint32_t, size_t>> lastRequest_;
  int64_t endOfUserRecordsOffset_{};
  uint32_t recordHeaderSize_{};
  bool fileHasAnIndex_{};
};
 
const IndexRecord::RecordInfo* getNearestRecordByTime(
    const std::vector<const IndexRecord::RecordInfo*>& index,
    double timestamp,
    double epsilon,
    Record::Type recordType = Record::Type::UNDEFINED);
 
} // namespace vrs