Project
Loading...
Searching...
No Matches
RootArrowFilesystem.h
Go to the documentation of this file.
1// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3// All rights not expressly granted are reserved.
4//
5// This software is distributed under the terms of the GNU General Public
6// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7//
8// In applying this license CERN does not waive the privileges and immunities
9// granted to it by virtue of its status as an Intergovernmental Organization
10// or submit itself to any jurisdiction.
11#ifndef O2_FRAMEWORK_ROOT_ARROW_FILESYSTEM_H_
12#define O2_FRAMEWORK_ROOT_ARROW_FILESYSTEM_H_
13
14#include <TBufferFile.h>
15#include <arrow/buffer.h>
16#include <arrow/dataset/dataset.h>
17#include <arrow/dataset/type_fwd.h>
18#include <arrow/dataset/file_base.h>
19#include <arrow/filesystem/type_fwd.h>
20#include <arrow/type_fwd.h>
21#include <memory>
22#include <utility>
23
24class TFile;
25class TBufferFile;
26class TDirectoryFile;
27
28namespace o2::framework
29{
30
32 RootObjectHandler(void* p, std::shared_ptr<arrow::dataset::FileFormat> f)
33 : payload(p), format(std::move(f))
34 {
35 }
36
37 ~RootObjectHandler() noexcept(false);
38
39 template <typename T>
40 std::unique_ptr<T> GetObjectAsOwner()
41 {
42 auto* p = payload;
43 payload = nullptr;
44 return std::unique_ptr<T>((T*)p);
45 }
46 std::shared_ptr<arrow::dataset::FileFormat> format;
47
48 private:
49 void* payload = nullptr;
50};
51
52// This is to avoid having to implement a bunch of unimplemented methods
53// for all the possible virtual filesystem we can invent on top of ROOT
54// data structures.
56{
57 public:
58 // Dummy implementation to avoid
59 arrow::Result<arrow::fs::FileInfo> GetFileInfo(const std::string& path) override;
60 arrow::Result<arrow::fs::FileInfoVector> GetFileInfo(const arrow::fs::FileSelector& select) override;
61
62 bool Equals(const FileSystem& other) const override
63 {
64 return this->type_name() == other.type_name();
65 }
66
67 virtual std::shared_ptr<RootObjectHandler> GetObjectHandler(arrow::dataset::FileSource source) = 0;
68 virtual bool CheckSupport(arrow::dataset::FileSource source) = 0;
69
70 arrow::Status CreateDir(const std::string& path, bool recursive) override;
71
72 arrow::Status DeleteDir(const std::string& path) override;
73
74 arrow::Status CopyFile(const std::string& src, const std::string& dest) override;
75
76 arrow::Status Move(const std::string& src, const std::string& dest) override;
77
78 arrow::Status DeleteDirContents(const std::string& path, bool missing_dir_ok) override;
79
80 arrow::Status DeleteRootDirContents() override;
81
82 arrow::Status DeleteFile(const std::string& path) override;
83
84 arrow::Result<std::shared_ptr<arrow::io::InputStream>> OpenInputStream(const std::string& path) override;
85
86 arrow::Result<std::shared_ptr<arrow::io::RandomAccessFile>> OpenInputFile(const std::string& path) override;
87
88 arrow::Result<std::shared_ptr<arrow::io::OutputStream>> OpenOutputStream(
89 const std::string& path,
90 const std::shared_ptr<const arrow::KeyValueMetadata>& metadata) override;
91
92 arrow::Result<std::shared_ptr<arrow::io::OutputStream>> OpenAppendStream(
93 const std::string& path,
94 const std::shared_ptr<const arrow::KeyValueMetadata>& metadata) override;
95};
96
97struct RootArrowFactory final {
98 std::function<std::shared_ptr<arrow::dataset::FileWriteOptions>()> options = nullptr;
99 std::function<std::shared_ptr<arrow::dataset::FileFormat>()> format = nullptr;
100 // Builds an output streamer which is able to read from the source fragment
101 // in a deferred way.
102 std::function<std::shared_ptr<arrow::io::OutputStream>(std::shared_ptr<arrow::dataset::FileFragment>, const std::shared_ptr<arrow::ResizableBuffer>& buffer)> deferredOutputStreamer = nullptr;
103};
104
108
109// A registry for all the possible ways of encoding a table in a TFile
111 // The unique name of this capability
112 std::string name = "unknown";
113 // Convert a logical filename to an actual object to be read
114 // This can be used, e.g. to read an RNTuple stored in
115 // a flat directory structure in a TFile vs a TTree stored inside
116 // a TDirectory (e.g. /DF_1000/o2tracks).
117 std::function<std::string(std::string)> lfn2objectPath;
118 // Given a TFile, return the object which this capability support
119 // Use a void * in order not to expose the kind of object to the
120 // generic reading code. This is also where we load the plugin
121 // which will be used for the actual creation.
122 std::function<void*(std::shared_ptr<arrow::fs::FileSystem> fs, std::string const& path)> getHandle;
123 // Wether or not this actually supports reading an object of the following class
124 std::function<bool(char const*)> checkSupport;
125
126 // Accounts the bytes of the object behind `handle` against the two counters, so that
127 // the generic reading code need not know how a given format reports its size. Left
128 // null by formats which cannot report it.
129 std::function<void(void* handle, size_t& compressed, size_t& uncompressed)> accountBytes = nullptr;
130
131 // This must be implemented to load the actual RootArrowFactory plugin which
132 // implements this capability. This way the detection of the file format
133 // (via get handle) does not need to know about the actual code which performs
134 // the serialization (and might depend on e.g. RNTuple).
135 std::function<RootArrowFactory&()> factory;
136};
137
141
142// This acts as registry of all the capabilities (i.e. the ability to
143// associate a given object in a root file to the serialization plugin) and
144// the factory (i.e. the serialization plugin)
146 std::vector<RootObjectReadingCapability> capabilities;
147};
148
150{
151 public:
152 arrow::Result<arrow::fs::FileInfo> GetFileInfo(const std::string& path) override;
153
154 TFileFileSystem(TDirectoryFile* f, size_t readahead, RootObjectReadingFactory&, bool ownsFile = true);
155
156 ~TFileFileSystem() override;
157
158 std::string type_name() const override
159 {
160 return "TDirectoryFile";
161 }
162
163 std::shared_ptr<RootObjectHandler> GetObjectHandler(arrow::dataset::FileSource source) override;
164 bool CheckSupport(arrow::dataset::FileSource source) override;
165 virtual std::shared_ptr<VirtualRootFileSystemBase> GetSubFilesystem(arrow::dataset::FileSource source);
166
167 arrow::Result<std::shared_ptr<arrow::io::OutputStream>> OpenOutputStream(
168 const std::string& path,
169 const std::shared_ptr<const arrow::KeyValueMetadata>& metadata) override;
170
171 // We can go back to the TFile in case this is needed.
172 TDirectoryFile* GetFile()
173 {
174 return mFile;
175 }
176
177 private:
178 TDirectoryFile* mFile;
179 RootObjectReadingFactory& mObjectFactory;
180 bool mOwnsFile = true;
181};
182
184{
185 public:
187
188 arrow::Result<arrow::fs::FileInfo> GetFileInfo(const std::string& path) override;
189 std::string type_name() const override
190 {
191 return "tbufferfile";
192 }
193
194 bool CheckSupport(arrow::dataset::FileSource source) override;
195 std::shared_ptr<RootObjectHandler> GetObjectHandler(arrow::dataset::FileSource source) override;
197 {
198 return mBuffer;
199 }
200
201 private:
202 TBufferFile* mBuffer;
203 std::shared_ptr<VirtualRootFileSystemBase> mFilesystem;
204 RootObjectReadingFactory& mObjectFactory;
205};
206
207// An arrow outputstream which allows to write to a TDirectoryFile.
208// This will point to the location of the file itself. You can
209// specify the location of the actual object inside it by passing the
210// associated path to the Write() API.
212{
213 public:
214 TDirectoryFileOutputStream(TDirectoryFile*);
215
216 arrow::Status Close() override;
217
218 arrow::Result<int64_t> Tell() const override;
219
220 arrow::Status Write(const void* data, int64_t nbytes) override;
221
222 bool closed() const override;
223
224 TDirectoryFile* GetDirectory()
225 {
226 return mDirectory;
227 }
228
229 private:
230 TDirectoryFile* mDirectory;
231};
232
233} // namespace o2::framework
234
235#endif // O2_FRAMEWORK_ROOT_ARROW_FILESYSTEM_H_
configStandalone Reconstruction rec_tpc multiplier for the combined cluster track error during track following area in cm for the search of for z only used if decrease removal by when above this local occupancy rowx Shift of TPC revert cluster flag edge bit distance to edge exceeds this ZOffset for secondary multiple of radius max P of electron K BB bands crossing max P of electron p BB bands crossing max P of electron d BB bands crossing max P of electron t BB bands crossing min accepted dEdxMaxTPC of the track Additive extra cluster error for Y2 if edge flag set Additive extra cluster error for Y2 if single set Additive extra cluster error for Y2 if or single set Additive extra cluster error for Z2 if or single set Extra errors to add to track when crossing depending on addErrorsCECrossing Inner portion of y range in sector that is not used in searching for extrapolated track candidates factor2YS for track merging factor2K for track merging Margin in cm of Y position when rejecting edge clusters based on uncorrected track Y Max fraction of shared clusters for tracklet Except pad from cluster finding if total number of charges in a fragment is above this Except pad from cluster finding if number of consecutive charges in a fragment is above this Threshold where a timebin is considered disabling the noisy pad check for that pad Treat signal above this minimum as saturated Smoothing factor for the exponential Highly Ionising Particle tail filter Number of timebins to use for the averaging Add additional custom track errors when crossing relative as fraction of Do N fit passes in final fit of< 0:reject at max -n hits") AddOptionRTC(rejectIFCLowRadiusCluster, uint8_t, 1, "", 0, "Reject clusters that get the IFC mask error during refit") AddOptionRTC(dEdxTruncLow, uint8_t, 2, "", 0, "Low truncation threshold, fraction of 128") AddOptionRTC(dEdxTruncHigh, uint8_t, 77, "", 0, "High truncation threshold, fraction of 128") AddOptionRTC(extrapolationTracking, int8_t, 1, "", 0, "Enable Extrapolation Tracking(prolong tracks to adjacent sectors to find short segments)") AddOptionRTC(disableRefitAttachment, uint8_t, 0, "", 0, "Bitmask to disable certain attachment steps during refit(1:attachment, 2:propagation, 4:loop following, 8:mirroring)") AddOptionRTC(rejectionStrategy, uint8_t, o2::gpu::GPUSettings::RejectionStrategyA, "", 0, "Enable rejection of TPC clusters for compression(0=no, 1=strategy A, 2=strategy B)") AddOptionRTC(mergeLoopersAfterburner, uint8_t, 1, "", 0, "Run afterburner for additional looper merging") AddOptionRTC(compressionTypeMask, uint8_t, o2::gpu::GPUSettings::CompressionFull, "", 0, "TPC Compression mode bits(1=truncate charge/width LSB, 2=differences, 4=track-model)") AddOptionRTC(compressionSortOrder, uint8_t, o2::gpu::GPUSettings::SortTime, "", 0, "Sort order of TPC compression(0=time, 1=pad, 2=Z-time-pad, 3=Z-pad-time, 4=no sorting(use incoming order))") AddOptionRTC(sigBitsCharge, uint8_t, 4, "", 0, "Number of significant bits for TPC cluster charge in compression mode 1") AddOptionRTC(sigBitsWidth, uint8_t, 3, "", 0, "Number of significant bits for TPC cluster width in compression mode 1") AddOptionRTC(dropLoopers, uint8_t, 0, "", 0, "Drop looping tracks starting from second loop") AddOptionRTC(mergerInterpolateErrors, uint8_t, 1, "", 0, "Use interpolation instead of extrapolation for chi2 based cluster rejection") AddOptionRTC(mergerInterpolateRejectAlsoOnCurrentPosition, uint8_t, 1, "", 0, "When using mergerInterpolateErrors, reject based on chi2 twice computed with interpolated and current track position starting from NDF > mergerNonInterpolateRejectMinNDF Merge tracks accross the central electrode Enable PID response Add sub threshold clusters in TPC dEdx computation OR mask of TPC flags that will reject the cluster in alternative dEdx Reject edge clusters based on uncorrected track Y during track fit Increase decrease removal by when on this number of pads from the edge h configStandalone Reconstruction rec_trd Max chi2 for TRD tracklets to be matched to a track Chi2 cut for strict matching mode Number of sigmas for ITS TPC track time error estimate Addition to search road around track prolongation along Y in cm Optimal tracklet rphi resolution in maximum chi2 per attached tracklet for TRD tracks take into account the full likelihood of the slope including exponential tails Tracks with less attached tracklets are discarded after the tracking Post trigger Pile up integration time in BCs fits all iterations but the last one in projections Use material lookup table for TPC refit Apply workarounds to avoid cyclic merge should not be needed tpc dyn configStandalone Processing proc_rtc Replace constant variables static by constexpr expressions false
configStandalone Reconstruction rec_tpc multiplier for the combined cluster track error during track following area in cm for the search of for z only used if decrease removal by when above this local occupancy rowx Shift of TPC revert cluster flag edge bit distance to edge exceeds this ZOffset for secondary multiple of radius max P of electron K BB bands crossing max P of electron p BB bands crossing max P of electron d BB bands crossing max P of electron t BB bands crossing min accepted dEdxMaxTPC of the track Additive extra cluster error for Y2 if edge flag set Additive extra cluster error for Y2 if single set Additive extra cluster error for Y2 if or single set Additive extra cluster error for Z2 if or single set Extra errors to add to track when crossing depending on addErrorsCECrossing Inner portion of y range in sector that is not used in searching for extrapolated track candidates factor2YS for track merging factor2K for track merging Margin in cm of Y position when rejecting edge clusters based on uncorrected track Y Max fraction of shared clusters for tracklet Except pad from cluster finding if total number of charges in a fragment is above this Except pad from cluster finding if number of consecutive charges in a fragment is above this Threshold where a timebin is considered disabling the noisy pad check for that pad Treat signal above this minimum as saturated Smoothing factor for the exponential Highly Ionising Particle tail filter Number of timebins to use for the averaging Add additional custom track errors when crossing relative as fraction of Do N fit passes in final fit of< 0:reject at max -n hits") AddOptionRTC(rejectIFCLowRadiusCluster, uint8_t, 1, "", 0, "Reject clusters that get the IFC mask error during refit") AddOptionRTC(dEdxTruncLow, uint8_t, 2, "", 0, "Low truncation threshold, fraction of 128") AddOptionRTC(dEdxTruncHigh, uint8_t, 77, "", 0, "High truncation threshold, fraction of 128") AddOptionRTC(extrapolationTracking, int8_t, 1, "", 0, "Enable Extrapolation Tracking(prolong tracks to adjacent sectors to find short segments)") AddOptionRTC(disableRefitAttachment, uint8_t, 0, "", 0, "Bitmask to disable certain attachment steps during refit(1:attachment, 2:propagation, 4:loop following, 8:mirroring)") AddOptionRTC(rejectionStrategy, uint8_t, o2::gpu::GPUSettings::RejectionStrategyA, "", 0, "Enable rejection of TPC clusters for compression(0=no, 1=strategy A, 2=strategy B)") AddOptionRTC(mergeLoopersAfterburner, uint8_t, 1, "", 0, "Run afterburner for additional looper merging") AddOptionRTC(compressionTypeMask, uint8_t, o2::gpu::GPUSettings::CompressionFull, "", 0, "TPC Compression mode bits(1=truncate charge/width LSB, 2=differences, 4=track-model)") AddOptionRTC(compressionSortOrder, uint8_t, o2::gpu::GPUSettings::SortTime, "", 0, "Sort order of TPC compression(0=time, 1=pad, 2=Z-time-pad, 3=Z-pad-time, 4=no sorting(use incoming order))") AddOptionRTC(sigBitsCharge, uint8_t, 4, "", 0, "Number of significant bits for TPC cluster charge in compression mode 1") AddOptionRTC(sigBitsWidth, uint8_t, 3, "", 0, "Number of significant bits for TPC cluster width in compression mode 1") AddOptionRTC(dropLoopers, uint8_t, 0, "", 0, "Drop looping tracks starting from second loop") AddOptionRTC(mergerInterpolateErrors, uint8_t, 1, "", 0, "Use interpolation instead of extrapolation for chi2 based cluster rejection") AddOptionRTC(mergerInterpolateRejectAlsoOnCurrentPosition, uint8_t, 1, "", 0, "When using mergerInterpolateErrors, reject based on chi2 twice computed with interpolated and current track position starting from NDF > mergerNonInterpolateRejectMinNDF Merge tracks accross the central electrode Enable PID response Add sub threshold clusters in TPC dEdx computation OR mask of TPC flags that will reject the cluster in alternative dEdx Reject edge clusters based on uncorrected track Y during track fit Increase decrease removal by when on this number of pads from the edge h configStandalone Reconstruction rec_trd Max chi2 for TRD tracklets to be matched to a track Chi2 cut for strict matching mode Number of sigmas for ITS TPC track time error estimate Addition to search road around track prolongation along Y in cm Optimal tracklet rphi resolution in maximum chi2 per attached tracklet for TRD tracks take into account the full likelihood of the slope including exponential tails Tracks with less attached tracklets are discarded after the tracking Post trigger Pile up integration time in BCs fits all iterations but the last one in projections Use material lookup table for TPC refit Apply workarounds to avoid cyclic merge should not be needed tpc dyn configStandalone Processing proc_rtc bool
CURL * handle
arrow::Result< arrow::fs::FileInfo > GetFileInfo(const std::string &path) override
std::shared_ptr< RootObjectHandler > GetObjectHandler(arrow::dataset::FileSource source) override
bool CheckSupport(arrow::dataset::FileSource source) override
std::string type_name() const override
arrow::Result< int64_t > Tell() const override
std::string type_name() const override
virtual std::shared_ptr< VirtualRootFileSystemBase > GetSubFilesystem(arrow::dataset::FileSource source)
bool CheckSupport(arrow::dataset::FileSource source) override
arrow::Result< arrow::fs::FileInfo > GetFileInfo(const std::string &path) override
arrow::Result< std::shared_ptr< arrow::io::OutputStream > > OpenOutputStream(const std::string &path, const std::shared_ptr< const arrow::KeyValueMetadata > &metadata) override
std::shared_ptr< RootObjectHandler > GetObjectHandler(arrow::dataset::FileSource source) override
arrow::Status DeleteDir(const std::string &path) override
virtual std::shared_ptr< RootObjectHandler > GetObjectHandler(arrow::dataset::FileSource source)=0
arrow::Status Move(const std::string &src, const std::string &dest) override
arrow::Status DeleteFile(const std::string &path) override
bool Equals(const FileSystem &other) const override
arrow::Status DeleteDirContents(const std::string &path, bool missing_dir_ok) override
arrow::Status CreateDir(const std::string &path, bool recursive) override
arrow::Status CopyFile(const std::string &src, const std::string &dest) override
arrow::Result< arrow::fs::FileInfo > GetFileInfo(const std::string &path) override
arrow::Result< std::shared_ptr< arrow::io::OutputStream > > OpenOutputStream(const std::string &path, const std::shared_ptr< const arrow::KeyValueMetadata > &metadata) override
arrow::Result< std::shared_ptr< arrow::io::InputStream > > OpenInputStream(const std::string &path) override
arrow::Result< std::shared_ptr< arrow::io::OutputStream > > OpenAppendStream(const std::string &path, const std::shared_ptr< const arrow::KeyValueMetadata > &metadata) override
virtual bool CheckSupport(arrow::dataset::FileSource source)=0
arrow::Result< std::shared_ptr< arrow::io::RandomAccessFile > > OpenInputFile(const std::string &path) override
GLuint const GLchar * name
Definition glcorearb.h:781
GLdouble f
Definition glcorearb.h:310
GLsizei GLsizei GLchar * source
Definition glcorearb.h:798
typedef void(APIENTRYP PFNGLCULLFACEPROC)(GLenum mode)
GLsizei const GLchar *const * path
Definition glcorearb.h:3591
GLint GLint GLsizei GLint GLenum format
Definition glcorearb.h:275
Defining ITS Vertex explicitly as messageable.
Definition Cartesian.h:288
std::vector< InputSpec > select(char const *matcher="")
virtual RootArrowFactory * create()=0
std::function< std::shared_ptr< arrow::dataset::FileWriteOptions >()> options
std::function< std::shared_ptr< arrow::io::OutputStream >(std::shared_ptr< arrow::dataset::FileFragment >, const std::shared_ptr< arrow::ResizableBuffer > &buffer)> deferredOutputStreamer
RootObjectHandler(void *p, std::shared_ptr< arrow::dataset::FileFormat > f)
std::unique_ptr< T > GetObjectAsOwner()
std::shared_ptr< arrow::dataset::FileFormat > format
virtual RootObjectReadingCapability * create()=0
std::function< RootArrowFactory &()> factory
std::function< void(void *handle, size_t &compressed, size_t &uncompressed)> accountBytes
std::function< bool(char const *)> checkSupport
std::function< std::string(std::string)> lfn2objectPath
std::function< void *(std::shared_ptr< arrow::fs::FileSystem > fs, std::string const &path)> getHandle
std::vector< RootObjectReadingCapability > capabilities
VectorOfTObjectPtrs other
ctfTree Write()