DPL Analysis: Expressions <-> JSON

Framework/Core/CMakeLists.txt

@@ -143,6 +143,7 @@ o2_add_library(Framework
                        src/Array2D.cxx
                        src/Variant.cxx
                        src/VariantJSONHelpers.cxx
+                       src/ExpressionJSONHelpers.cxx
                        src/VariantPropertyTreeHelpers.cxx
                        src/WorkflowCustomizationHelpers.cxx
                        src/WorkflowHelpers.cxx

Framework/Core/include/Framework/ASoA.h

@@ -1293,7 +1293,14 @@ concept with_ccdb_urls = requires {
 };
 
 template <typename T>
-concept with_base_table = not_void<typename aod::MetadataTrait<o2::aod::Hash<T::ref.desc_hash>>::metadata::base_table_t>;
+concept with_base_table = requires {
+  typename aod::MetadataTrait<o2::aod::Hash<T::ref.desc_hash>>::metadata::base_table_t;
+};
+
+template <typename T>
+concept with_expression_pack = requires {
+  typename T::expression_pack_t{};
+};
 
 template <size_t N1, std::array<TableRef, N1> os1, size_t N2, std::array<TableRef, N2> os2>
 consteval bool is_compatible()

Framework/Core/include/Framework/AnalysisHelpers.h

@@ -26,6 +26,11 @@
 #include "Framework/Traits.h"
 
 #include <string>
+namespace o2::framework {
+std::string serializeProjectors(std::vector<framework::expressions::Projector>& projectors);
+std::string serializeSchema(std::shared_ptr<arrow::Schema>& schema);
+}
+
 namespace o2::soa
 {
 template <TableRef R>
@@ -97,6 +102,32 @@ constexpr auto getCCDBMetadata() -> std::vector<framework::ConfigParamSpec>
 {
   return {};
 }
+
+template <soa::with_expression_pack T>
+constexpr auto getExpressionMetadata() -> std::vector<framework::ConfigParamSpec>
+{
+  using expression_pack_t = T::expression_pack_t;
+
+  auto projectors = []<typename... C>(framework::pack<C...>) -> std::vector<framework::expressions::Projector> {
+    std::vector<framework::expressions::Projector> result;
+    (result.emplace_back(std::move(C::Projector())), ...);
+    return result;
+  }(expression_pack_t{});
+
+  auto schema = std::make_shared<arrow::Schema>(o2::soa::createFieldsFromColumns(expression_pack_t{}));
+
+  auto json = framework::serializeProjectors(projectors);
+  return {framework::ConfigParamSpec{"projectors", framework::VariantType::String, json, {"\"\""}},
+          framework::ConfigParamSpec{"schema", framework::VariantType::String, framework::serializeSchema(schema), {"\"\""}}};
+}
+
+template <typename T>
+  requires(!soa::with_expression_pack<T>)
+constexpr auto getExpressionMetadata() -> std::vector<framework::ConfigParamSpec>
+{
+  return {};
+}
+
 }  // namespace
 
 template <TableRef R>
@@ -107,6 +138,8 @@ constexpr auto tableRef2InputSpec()
   metadata.insert(metadata.end(), m.begin(), m.end());
   auto ccdbMetadata = getCCDBMetadata<typename o2::aod::MetadataTrait<o2::aod::Hash<R.desc_hash>>::metadata>();
   metadata.insert(metadata.end(), ccdbMetadata.begin(), ccdbMetadata.end());
+  auto p = getExpressionMetadata<typename o2::aod::MetadataTrait<o2::aod::Hash<R.desc_hash>>::metadata>();
+  metadata.insert(metadata.end(), p.begin(), p.end());
 
   return framework::InputSpec{
     o2::aod::label<R>(),

Framework/Core/include/Framework/Expressions.h

@@ -110,6 +110,8 @@ std::string upcastTo(atype::type f);
 
 /// An expression tree node corresponding to a literal value
 struct LiteralNode {
+  using var_t = LiteralValue::stored_type;
+
   LiteralNode()
     : value{-1},
       type{atype::INT32}
@@ -120,7 +122,12 @@ struct LiteralNode {
   {
   }
 
-  using var_t = LiteralValue::stored_type;
+  LiteralNode(var_t v, atype::type t)
+    : value{v},
+      type{t}
+  {
+  }
+
   var_t value;
   atype::type type = atype::NA;
 };
@@ -617,14 +624,19 @@ inline Node ncfg(T defaultValue, std::string path)
 struct Filter {
   Filter() = default;
 
+  Filter(std::unique_ptr<Node>&& ptr)
+  {
+    node = std::move(ptr);
+    (void)designateSubtrees(node.get());
+  }
+
   Filter(Node&& node_) : node{std::make_unique<Node>(std::forward<Node>(node_))}
   {
     (void)designateSubtrees(node.get());
   }
 
   Filter(Filter&& other) : node{std::forward<std::unique_ptr<Node>>(other.node)}
   {
-    (void)designateSubtrees(node.get());
   }
 
   Filter(std::string const& input_) : input{input_} {}

Framework/Core/src/AODReaderHelpers.cxx

@@ -19,6 +19,7 @@
 #include "Framework/CallbackService.h"
 #include "Framework/EndOfStreamContext.h"
 #include "Framework/DataSpecUtils.h"
+#include "Framework/ExpressionJSONHelpers.h"
 
 #include <Monitoring/Monitoring.h>
 
@@ -44,28 +45,6 @@ auto setEOSCallback(InitContext& ic)
     });
 }
 
-template <typename... Ts>
-static inline auto doExtractOriginal(framework::pack<Ts...>, ProcessingContext& pc)
-{
-  if constexpr (sizeof...(Ts) == 1) {
-    return pc.inputs().get<TableConsumer>(aod::MetadataTrait<framework::pack_element_t<0, framework::pack<Ts...>>>::metadata::tableLabel())->asArrowTable();
-  } else {
-    return std::vector{pc.inputs().get<TableConsumer>(aod::MetadataTrait<Ts>::metadata::tableLabel())->asArrowTable()...};
-  }
-}
-
-template <typename... Os>
-static inline auto extractOriginalsTuple(framework::pack<Os...>, ProcessingContext& pc)
-{
-  return std::make_tuple(extractTypedOriginal<Os>(pc)...);
-}
-
-template <typename... Os>
-static inline auto extractOriginalsVector(framework::pack<Os...>, ProcessingContext& pc)
-{
-  return std::vector{extractOriginal<Os>(pc)...};
-}
-
 template <size_t N, std::array<soa::TableRef, N> refs>
 static inline auto extractOriginals(ProcessingContext& pc)
 {
@@ -156,12 +135,32 @@ auto make_spawn(InputSpec const& input, ProcessingContext& pc)
   (typename metadata_t::expression_pack_t{});
   return o2::framework::spawner<D>(extractOriginals<sources.size(), sources>(pc), input.binding.c_str(), projectors.data(), projector, schema);
 }
+
+struct Spawnable {
+  std::vector<expressions::Projector> projectors;
+  std::vector<std::string> labels;
+  std::shared_ptr<arrow::Schema> schema;
+
+  Spawnable(InputSpec const& spec)
+  {
+    auto loc = std::find_if(spec.metadata.begin(), spec.metadata.end(), [](ConfigParamSpec const& spc){ return spc.name.compare("projectors") == 0; });
+    std::stringstream iws(loc->defaultValue.get<std::string>());
+    projectors = ExpressionJSONHelpers::read(iws);
+    for (auto& i : spec.metadata) {
+      if (i.name.starts_with("input:")) {
+        labels.emplace_back(i.name.substr(6));
+      }
+    }
+  }
+};
 } // namespace
 
 AlgorithmSpec AODReaderHelpers::aodSpawnerCallback(std::vector<InputSpec>& requested)
 {
   return AlgorithmSpec::InitCallback{[requested](InitContext& /*ic*/) {
-    return [requested](ProcessingContext& pc) {
+    std::vector<Spawnable> spawnables;
+
+    return [requested, spawnables](ProcessingContext& pc) {
       auto outputs = pc.outputs();
       // spawn tables
       for (auto& input : requested) {

Framework/Core/src/AnalysisHelpers.cxx

@@ -9,6 +9,7 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 #include "Framework/ExpressionHelpers.h"
+#include "ExpressionJSONHelpers.h"
 
 namespace o2::framework
 {
@@ -26,4 +27,18 @@ void initializePartitionCaches(std::set<uint32_t> const& hashes, std::shared_ptr
     gfilter = framework::expressions::createFilter(schema, framework::expressions::makeCondition(tree));
   }
 }
+
+std::string serializeProjectors(std::vector<framework::expressions::Projector>& projectors)
+{
+  std::stringstream osm;
+  ExpressionJSONHelpers::write(osm, projectors);
+  return osm.str();
+}
+
+std::string serializeSchema(std::shared_ptr<arrow::Schema>& schema)
+{
+  std::stringstream osm;
+  ArrowJSONHelpers::write(osm, schema);
+  return osm.str();
+}
 } // namespace o2::framework