Track skipped read timeframes in file statistics & dumping (#15735)

* Track skipped read timeframes in file statistics, dump

* Use concrete when recording skipped timeframe

Author: autumn-mck

Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx

@@ -271,10 +271,11 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
 
       int64_t startTime = uv_hrtime();
       int64_t startSize = totalSizeCompressed;
-      auto skipInvalidRead = [&](o2::header::DataOrigin const& origin, InvalidAODReadError const& e) {
+      auto skipInvalidRead = [&](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) {
         auto skippedTimeframes = ++totalInvalidReadSkipped;
         LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
-             origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
+             concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
+        didir->markTimeFrameSkipped(header::DataHeader(concrete.description, concrete.origin, concrete.subSpec), ntf);
         arrowContext.clear();
         messageContext.discard();
         stringContext.clear();
@@ -361,7 +362,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
           if (!skipInvalidReads) {
             throw;
           }
-          skipInvalidRead(concrete.origin, e);
+          skipInvalidRead(concrete, e);
           return TFReaderState::INVALID_TIMEFRAME;
         }
 

Framework/AnalysisSupport/src/DataInputDirector.cxx

@@ -288,6 +288,15 @@ arrow::dataset::FileSource DataInputDescriptor::getFileFolder(int counter, int n
   return {fmt::format("DF_{}", mfilenames[counter].listOfTimeFrameNumbers[numTF]), mCurrentFilesystem};
 }
 
+uint64_t DataInputDescriptor::markTimeFrameSkipped(int numTF)
+{
+  if (mCurrentFileID >= 0 && numTF >= 0 && numTF < mfilenames[mCurrentFileID].numberOfTimeFrames) {
+    mfilenames[mCurrentFileID].alreadyRead[numTF] = false;
+    return ++mfilenames[mCurrentFileID].invalidReadSkipped;
+  }
+  return 0;
+}
+
 std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int counter, int numTF, std::string treename, int wantedParentLevel, std::string_view wantedOrigin)
 {
   if (!mParentFileMap) {
@@ -364,8 +373,8 @@ void DataInputDescriptor::printFileStatistics()
   }
   auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
   auto f = dynamic_cast<TFile*>(rootFS->GetFile());
-  std::string monitoringInfo(fmt::format("lfn={},size={},total_df={},read_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}", f->GetName(),
-                                         f->GetSize(), getTimeFramesInFile(mCurrentFileID), getReadTimeFramesInFile(mCurrentFileID), f->GetBytesRead(), f->GetReadCalls(),
+  std::string monitoringInfo(fmt::format("lfn={},size={},total_df={},read_df={},skipped_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}", f->GetName(),
+                                         f->GetSize(), getTimeFramesInFile(mCurrentFileID), getReadTimeFramesInFile(mCurrentFileID), mfilenames.at(mCurrentFileID).invalidReadSkipped, f->GetBytesRead(), f->GetReadCalls(),
                                          ((float)mIOTime / 1e9), ((float)wait_time / 1e9), mLevel));
 #if __has_include(<TJAlienFile.h>)
   auto alienFile = dynamic_cast<TJAlienFile*>(f);
@@ -879,6 +888,15 @@ arrow::dataset::FileSource DataInputDirector::getFileFolder(header::DataHeader d
   return didesc->getFileFolder(counter, numTF, wantedLevel, origin);
 }
 
+void DataInputDirector::markTimeFrameSkipped(header::DataHeader dh, int numTF)
+{
+  auto didesc = getDataInputDescriptor(dh);
+  if (!didesc) {
+    didesc = mdefaultDataInputDescriptor.get();
+  }
+  didesc->markTimeFrameSkipped(numTF);
+}
+
 int DataInputDirector::getTimeFramesInFile(header::DataHeader dh, int counter)
 {
   auto didesc = getDataInputDescriptor(dh);

Framework/AnalysisSupport/src/DataInputDirector.h

@@ -44,6 +44,7 @@ struct FileNameHolder {
   int numberOfTimeFrames = 0;
   std::vector<uint64_t> listOfTimeFrameNumbers;
   std::vector<bool> alreadyRead;
+  uint64_t invalidReadSkipped = 0;
 };
 
 FileNameHolder makeFileNameHolder(std::string fileName);
@@ -106,6 +107,7 @@ class DataInputDescriptor
 
   uint64_t getTimeFrameNumber(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin);
   arrow::dataset::FileSource getFileFolder(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin);
+  uint64_t markTimeFrameSkipped(int numTF);
   // Open the current file to populate the parent map, then return the parent descriptor and
   // the TF index within it that corresponds to numTF at this level. Returns {nullptr, -1} on failure.
   std::pair<std::shared_ptr<DataInputDescriptor>, int> navigateToLevel(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin);
@@ -171,6 +173,7 @@ class DataInputDirector
   bool readTree(DataAllocator& outputs, header::DataHeader dh, int counter, int numTF, size_t& totalSizeCompressed, size_t& totalSizeUncompressed, bool wasAOD);
   uint64_t getTimeFrameNumber(header::DataHeader dh, int counter, int numTF);
   arrow::dataset::FileSource getFileFolder(header::DataHeader dh, int counter, int numTF);
+  void markTimeFrameSkipped(header::DataHeader dh, int numTF);
   int getTimeFramesInFile(header::DataHeader dh, int counter);
 
   uint64_t getTotalSizeCompressed();

Framework/Core/src/runDataProcessing.cxx

@@ -1253,6 +1253,7 @@ std::vector<std::regex> getDumpableMetrics()
   dumpableMetrics.emplace_back("^aod-bytes-read-compressed$");
   dumpableMetrics.emplace_back("^aod-file-read-info$");
   dumpableMetrics.emplace_back("^aod-largest-object-written$");
+  dumpableMetrics.emplace_back("^aod-invalid-read-skipped-timeframes$");
   dumpableMetrics.emplace_back("^table-bytes-.*");
   dumpableMetrics.emplace_back("^total-timeframes.*");
   dumpableMetrics.emplace_back("^device_state.*");