Merge 13071e7 into sapling-pr-archive-ktf

Author: ktf

Framework/Core/include/Framework/ASoA.h

@@ -2308,6 +2308,28 @@ class Table
 template <uint32_t D, soa::is_column... C>
 using InPlaceTable = Table<o2::aod::Hash<"TEST"_h>, o2::aod::Hash<D>, o2::aod::Hash<"TEST"_h>, C...>;
 
+template <typename Narrow>
+std::shared_ptr<arrow::Table> projectColumns(std::shared_ptr<arrow::Table> const& src)
+{
+  auto indices = []<typename... C>(framework::pack<C...>, std::shared_ptr<arrow::Schema> const& sch) {
+    return std::vector<int>{sch->GetFieldIndex(C::columnLabel())...};
+  }(typename Narrow::persistent_columns_t{}, src->schema());
+  return src->SelectColumns(indices).ValueOrDie();
+}
+
+template <typename Narrow, is_table T>
+  requires(!is_filtered_table<T>)
+Narrow as_table(T const& wide)
+{
+  return Narrow{projectColumns<Narrow>(wide.asArrowTable()), wide.offset()};
+}
+
+template <typename Narrow, is_filtered_table T>
+soa::Filtered<Narrow> as_table(T const& wide)
+{
+  return soa::Filtered<Narrow>({projectColumns<Narrow>(wide.asArrowTable())}, wide.getSelectedRows(), wide.offset());
+}
+
 void getterNotFound(const char* targetColumnLabel);
 void emptyColumnLabel();
 

Framework/Core/test/test_ASoA.cxx

@@ -320,6 +320,70 @@ TEST_CASE("TestJoinedTables")
   }
 }
 
+TEST_CASE("TestAsTableProjection")
+{
+  TableBuilder builderX;
+  auto rowWriterX = builderX.persist<int32_t>({"fX"});
+  for (int i = 0; i < 8; ++i) {
+    rowWriterX(0, i);
+  }
+  auto tableX = builderX.finalize();
+
+  TableBuilder builderY;
+  auto rowWriterY = builderY.persist<int32_t>({"fY"});
+  for (int i = 0; i < 8; ++i) {
+    rowWriterY(0, 7 - i);
+  }
+  auto tableY = builderY.finalize();
+
+  TableBuilder builderZ;
+  auto rowWriterZ = builderZ.persist<int32_t>({"fZ"});
+  for (int i = 0; i < 8; ++i) {
+    rowWriterZ(0, 8);
+  }
+  auto tableZ = builderZ.finalize();
+
+  using TestX = InPlaceTable<"A0"_h, o2::aod::test::X>;
+  using TestY = InPlaceTable<"A1"_h, o2::aod::test::Y>;
+  using TestZ = InPlaceTable<"A2"_h, o2::aod::test::Z>;
+  using Wide = Join<TestX, TestY, TestZ>;
+
+  Wide wide{{tableX, tableY, tableZ}, 0};
+  REQUIRE(wide.asArrowTable()->num_columns() == 3);
+
+  // Project the wide join onto just the X table: the resulting arrow table
+  // shares the data but exposes a single column, and iteration sees only x.
+  auto narrow = as_table<TestX>(wide);
+  REQUIRE(narrow.asArrowTable()->num_columns() == 1);
+  REQUIRE(narrow.size() == 8);
+  int sum = 0;
+  for (auto& r : narrow) {
+    sum += r.x();
+  }
+  REQUIRE(sum == (0 + 1 + 2 + 3 + 4 + 5 + 6 + 7));
+
+  // Projecting onto two of the three tables keeps two columns.
+  auto narrow2 = as_table<Join<TestX, TestY>>(wide);
+  REQUIRE(narrow2.asArrowTable()->num_columns() == 2);
+  for (auto& r : narrow2) {
+    REQUIRE(7 == r.x() + r.y());
+  }
+
+  // Filtered projection: the selection carries over to the narrow table, so the
+  // narrow view iterates exactly the selected rows of the projected column.
+  Filtered<Wide> wideF{{tableX, tableY, tableZ}, SelectionVector{1, 3, 5}, 0};
+  auto narrowF = as_table<TestX>(wideF);
+  REQUIRE(narrowF.asArrowTable()->num_columns() == 1);
+  int sumF = 0;
+  int nF = 0;
+  for (auto& r : narrowF) {
+    sumF += r.x();
+    ++nF;
+  }
+  REQUIRE(nF == 3);
+  REQUIRE(sumF == (1 + 3 + 5));
+}
+
 TEST_CASE("TestConcatTables")
 {
   TableBuilder builderA;

GPU/GPUTracking/DataCompression/GPUTPCClusterStatistics.cxx

@@ -19,6 +19,7 @@
 #include <cstring>
 #include <map>
 #include <queue>
+#include <fstream>
 
 using namespace o2::gpu;
 
@@ -106,7 +107,7 @@ void GenerateCodes(const INode* node, const HuffCode& prefix, HuffCodeMap& outCo
 } // anonymous namespace
 } // namespace o2::gpu
 
-void GPUTPCClusterStatistics::RunStatistics(const o2::tpc::ClusterNativeAccess* clustersNative, const o2::tpc::CompressedClusters* clustersCompressed, const GPUParam& param)
+void GPUTPCClusterStatistics::RunStatistics(const o2::tpc::ClusterNativeAccess* clustersNative, const o2::tpc::CompressedClusters* clustersCompressed, const GPUParam& param, bool dumpCSV)
 {
   uint32_t decodingErrors = 0;
   o2::tpc::ClusterNativeAccess clustersNativeDecoded;
@@ -185,6 +186,49 @@ void GPUTPCClusterStatistics::RunStatistics(const o2::tpc::ClusterNativeAccess*
   FillStatisticCombined(mPQU, clustersCompressed->qMaxU, clustersCompressed->qTotU, clustersCompressed->nUnattachedClusters, P_MAX_QMAX);
   FillStatisticCombined(mProwSectorA, clustersCompressed->rowDiffA, clustersCompressed->sliceLegDiffA, clustersCompressed->nAttachedClustersReduced, GPUTPCGeometry::NROWS);
   mNTotalClusters += clustersCompressed->nAttachedClusters + clustersCompressed->nUnattachedClusters;
+
+  if (dumpCSV) {
+    std::ofstream csv("clusters_raw.csv");
+    csv << "sector,row,time,pad,flags,qtot,qmax,sigmatime,sigmapad\n";
+    for (uint32_t i = 0; i < NSECTORS; i++) {
+      for (uint32_t j = 0; j < GPUTPCGeometry::NROWS; j++) {
+        for (uint32_t k = 0; k < clustersNativeDecoded.nClusters[i][j]; k++) {
+          const auto& cl = clustersNativeDecoded.clusters[i][j][k];
+          csv << i << ',' << j << ',' << cl.getTimePacked() << ',' << cl.padPacked << ',' << (uint32_t)cl.getFlags() << ',' << cl.qTot << ',' << cl.qMax << ',' << (uint32_t)cl.sigmaTimePacked << ',' << (uint32_t)cl.sigmaPadPacked << '\n';
+        }
+      }
+    }
+
+    csv = std::ofstream("attachedCl.csv");
+    csv << "qTotA,qMaxA,flagsA,sigmaPadA,sigmaTimeA\n";
+    for (uint32_t i = 0; i < clustersCompressed->nAttachedClusters; i++) {
+      csv << clustersCompressed->qTotA[i] << ',' << clustersCompressed->qMaxA[i] << ',' << (uint32_t)clustersCompressed->flagsA[i] << ',' << (uint32_t)clustersCompressed->sigmaPadA[i] << ',' << (uint32_t)clustersCompressed->sigmaTimeA[i] << "\n";
+    }
+
+    csv = std::ofstream("attachedClred.csv");
+    csv << "rodDiffA,legDiffA,padResA,timeResA\n";
+    for (uint32_t i = 0; i < clustersCompressed->nAttachedClustersReduced; i++) {
+      csv << (uint32_t)clustersCompressed->rowDiffA[i] << ',' << (uint32_t)clustersCompressed->sliceLegDiffA[i] << ',' << clustersCompressed->padResA[i] << ',' << clustersCompressed->timeResA[i] << "\n";
+    }
+
+    csv = std::ofstream("nClU.csv");
+    csv << "sliceRowCl\n";
+    for (uint32_t i = 0; i < clustersCompressed->nSliceRows; i++) {
+      csv << clustersCompressed->nSliceRowClusters[i] << "\n";
+    }
+
+    csv = std::ofstream("trk.csv");
+    csv << "qPtA,rowA,sliceA,timeA,padA,nCl\n";
+    for (uint32_t i = 0; i < clustersCompressed->nTracks; i++) {
+      csv << (uint32_t)clustersCompressed->qPtA[i] << ',' << (uint32_t)clustersCompressed->rowA[i] << ',' << (uint32_t)clustersCompressed->sliceA[i] << ',' << clustersCompressed->timeA[i] << ',' << clustersCompressed->padA[i] << ',' << clustersCompressed->nTrackClusters[i] << "\n";
+    }
+
+    csv = std::ofstream("unattachedCl.csv");
+    csv << "qTotU,qMaxU,flagsU,padDiffU,timeDiffU,sigmaPadU,sigmaTimeU\n";
+    for (uint32_t i = 0; i < clustersCompressed->nUnattachedClusters; i++) {
+      csv << clustersCompressed->qTotU[i] << ',' << clustersCompressed->qMaxU[i] << ',' << (uint32_t)clustersCompressed->flagsU[i] << ',' << clustersCompressed->padDiffU[i] << ',' << clustersCompressed->timeDiffU[i] << ',' << (uint32_t)clustersCompressed->sigmaPadU[i] << ',' << (uint32_t)clustersCompressed->sigmaTimeU[i] << "\n";
+    }
+  }
 }
 
 void GPUTPCClusterStatistics::Finish()

GPU/GPUTracking/DataCompression/GPUTPCClusterStatistics.h

@@ -30,7 +30,7 @@ class GPUTPCClusterStatistics
 {
  public:
   static constexpr uint32_t NSECTORS = GPUTPCGeometry::NSECTORS;
-  void RunStatistics(const o2::tpc::ClusterNativeAccess* clustersNative, const o2::tpc::CompressedClusters* clustersCompressed, const GPUParam& param);
+  void RunStatistics(const o2::tpc::ClusterNativeAccess* clustersNative, const o2::tpc::CompressedClusters* clustersCompressed, const GPUParam& param, bool dumpCSV);
   void Finish();
 
  protected:

GPU/GPUTracking/Definitions/GPUSettingsList.h

@@ -342,7 +342,7 @@ AddOption(serializeGPU, int8_t, 0, "", 0, "Synchronize after each kernel call (b
 AddOption(recoTaskTiming, bool, 0, "", 0, "Perform summary timing after whole reconstruction tasks")
 AddOption(deterministicGPUReconstruction, int32_t, -1, "", 0, "Make CPU and GPU debug output comparable (sort / skip concurrent parts), -1 = automatic if debugLevel >= 6 or deterministic compile flag set", def(1))
 AddOption(showOutputStat, bool, false, "", 0, "Print some track output statistics")
-AddOption(runCompressionStatistics, bool, false, "compressionStat", 0, "Run statistics and verification for cluster compression")
+AddOption(runCompressionStatistics, int8_t, 0, "compressionStat", 0, "Run statistics and verification for cluster compression, 2 to dump clusters and entropy-reduced clusters to CSV", def(1))
 AddOption(resetTimers, int8_t, 1, "", 0, "Reset timers every event")
 AddOption(deviceTimers, bool, true, "", 0, "Use device timers instead of host-based time measurement")
 AddOption(keepAllMemory, bool, false, "", 0, "Allocate all memory on both device and host, and do not reuse")

GPU/GPUTracking/Global/GPUChainTracking.cxx

@@ -789,7 +789,7 @@ int32_t GPUChainTracking::RunChainFinalize()
 {
   if (mIOPtrs.clustersNative && (GetRecoSteps() & RecoStep::TPCCompression) && GetProcessingSettings().runCompressionStatistics) {
     CompressedClusters c = *mIOPtrs.tpcCompressedClusters;
-    mCompressionStatistics->RunStatistics(mIOPtrs.clustersNative, &c, param());
+    mCompressionStatistics->RunStatistics(mIOPtrs.clustersNative, &c, param(), GetProcessingSettings().runCompressionStatistics >= 2);
   }
 
   if (GetProcessingSettings().outputSanityCheck) {