DPL: add support for decompressing directly to shared memory

This PR postpones the read operations which would usually populate an intermediate
RecordBatch and it performs them directly on its subsequent shared memory
serialization. Doing so avoids having the intermediate representation allocate most
of the memory.

For the moment this is only done for TTree. RNtuple support will come in a subsequent PR.

Author: ktf

Framework/AnalysisSupport/src/DataInputDirector.cxx

@@ -399,6 +399,7 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh
   }
   // FIXME: Ugly. We should detect the format from the treename, good enough for now.
   std::shared_ptr<arrow::dataset::FileFormat> format;
+  FragmentToBatch::StreamerCreator creator = nullptr;
 
   auto fullpath = arrow::dataset::FileSource{folder.path() + "/" + treename, folder.filesystem()};
 
@@ -407,6 +408,7 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh
     void* handle = capability.getHandle(rootFS, objectPath);
     if (handle) {
       format = capability.factory().format();
+      creator = capability.factory().deferredOutputStreamer;
       break;
     }
   }
@@ -449,13 +451,12 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh
 
   // FIXME: This should allow me to create a memory pool
   // which I can then use to scan the dataset.
-  //
-  auto f2b = outputs.make<FragmentToBatch>(o);
+  auto f2b = outputs.make<FragmentToBatch>(o, creator, *fragment);
 
   //// add branches to read
   //// fill the table
   f2b->setLabel(treename.c_str());
-  f2b->fill(*fragment, datasetSchema, format);
+  f2b->fill(datasetSchema, format);
 
   mIOTime += (uv_hrtime() - ioStart);
 

Framework/AnalysisSupport/src/RNTuplePlugin.cxx

@@ -12,6 +12,7 @@
 #include "Framework/RuntimeError.h"
 #include "Framework/RootArrowFilesystem.h"
 #include "Framework/Plugins.h"
+#include "Framework/FairMQResizableBuffer.h"
 #include <ROOT/RNTupleModel.hxx>
 #include <ROOT/RNTupleWriteOptions.hxx>
 #include <ROOT/RNTupleWriter.hxx>
@@ -852,7 +853,10 @@ struct RNTupleObjectReadingImplementation : public RootArrowFactoryPlugin {
     return new RootArrowFactory{
       .options = [context]() { return context->format->DefaultWriteOptions(); },
       .format = [context]() { return context->format; },
-    };
+      .deferredOutputStreamer = [](std::shared_ptr<arrow::dataset::FileFragment> fragment, const std::shared_ptr<arrow::ResizableBuffer>& buffer) -> std::shared_ptr<arrow::io::OutputStream> {
+        auto treeFragment = std::dynamic_pointer_cast<RNTupleFileFragment>(fragment);
+        return std::make_shared<FairMQOutputStream>(buffer);
+      }};
   }
 };