Add FORM output module for phlex product_store integration (#177)

* Add FORM output module for phlex product persistence

Implements FormOutputModule that integrates phlex's product_store with
FORM's persistence layer. The module extracts products from phlex stores
and writes them using configurable FORM backend technologies.

Key changes:
- form_module.cpp: New output module with phlex product_store integration
- Updates to form/CMakeLists.txt to build form_module library
- Links against phlex::module and phlex::experimental interfaces

Note: Requires phlex standalone build modifications (submitted separately)

* Change to form::experimental::product_with_name

* Add test_helpers.hpp with createTypeMap implementation for FORM tests

* Remove mock_phlex directory

* Add integration test for form_module plugin. This test exercises FormOutputModule by running a Phlex workflow that uses the form_module plugin, improving code coverage for form_module.cpp. The test follows the same pattern as test/plugins, using a jsonnet configuration file to instantiate and run the module through Phlex.

* form: Properly store the TTree in its directory

* Implementing Peter's suggestion: switch from sprintf to snprintf

* Form: register more supported type.

This is a temporary work-around until std::type_info is part of the Phlex API.
Adding support for:
  long, float, double
and
  std::vector of int, long, float and double.

* Read product names from the jsonnet config 'products' field, making it generic for any configured products.

---------

Co-authored-by: Kyle Knoepfel <knoepfel@fnal.gov>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Philippe Canal <pcanal@fnal.gov>

Author: 4 people

form/CMakeLists.txt

@@ -8,6 +8,8 @@
 # ##############################################################################
 # Copyright (C) 2025 ...
 
+set(CMAKE_POSITION_INDEPENDENT_CODE ON)
+
 include_directories(${PROJECT_SOURCE_DIR}/form)
 
 # ROOT Storage toggle
@@ -18,11 +20,16 @@ endif()
 
 # Add sub directories
 add_subdirectory(form)
-add_subdirectory(mock_phlex)
 add_subdirectory(core)
 add_subdirectory(util)
 add_subdirectory(persistence)
 add_subdirectory(storage)
 if(FORM_USE_ROOT_STORAGE)
   add_subdirectory(root_storage)
 endif()
+
+add_library(form_module MODULE form_module.cpp)
+
+target_link_libraries(form_module PRIVATE phlex::module form)
+
+target_include_directories(form_module PRIVATE ${PROJECT_SOURCE_DIR})

form/form/form.cpp

@@ -4,88 +4,85 @@
 
 namespace form::experimental {
 
-  // Accept and store config
-  form_interface::form_interface(std::shared_ptr<mock_phlex::product_type_names> tm,
-                                 mock_phlex::config::parse_config const& config) :
+  form_interface::form_interface(std::shared_ptr<product_type_names> tm,
+                                 config::output_item_config const& output_config,
+                                 config::tech_setting_config const& tech_config) :
     m_pers(nullptr), m_type_map(tm)
   {
-    // Convert phlex config to form config
-    form::experimental::config::output_item_config output_items;
-    for (auto const& phlex_item : config.getItems()) {
-      output_items.addItem(phlex_item.product_name, phlex_item.file_name, phlex_item.technology);
-      m_product_to_config.emplace(
-        phlex_item.product_name,
-        form::experimental::config::PersistenceItem(
-          phlex_item.product_name, phlex_item.file_name, phlex_item.technology));
+    for (auto const& item : output_config.getItems()) {
+      m_product_to_config.emplace(item.product_name,
+                                  form::experimental::config::PersistenceItem(
+                                    item.product_name, item.file_name, item.technology));
     }
 
-    config::tech_setting_config tech_config_settings;
-    tech_config_settings.file_settings = config.getFileSettings();
-    tech_config_settings.container_settings = config.getContainerSettings();
-
     m_pers = form::detail::experimental::createPersistence();
-    m_pers->configureOutputItems(output_items);
-    m_pers->configureTechSettings(tech_config_settings);
+    m_pers->configureOutputItems(output_config);
+    m_pers->configureTechSettings(tech_config);
   }
 
-  void form_interface::write(std::string const& creator, mock_phlex::product_base const& pb)
+  void form_interface::write(std::string const& creator,
+                             std::string const& segment_id,
+                             product_with_name const& pb)
   {
-    // Look up creator from PersistenceItem.
+
     auto it = m_product_to_config.find(pb.label);
     if (it == m_product_to_config.end()) {
       throw std::runtime_error("No configuration found for product: " + pb.label);
     }
 
     std::string const type = m_type_map->names[pb.type];
-    // FIXME: Really only needed on first call
+
     std::map<std::string, std::string> products = {{pb.label, type}};
     m_pers->createContainers(creator, products);
+
     m_pers->registerWrite(creator, pb.label, pb.data, type);
-    m_pers->commitOutput(creator, pb.id);
+
+    m_pers->commitOutput(creator, segment_id);
   }
 
-  // Look up creator from config
   void form_interface::write(std::string const& creator,
-                             std::vector<mock_phlex::product_base> const& batch)
+                             std::string const& segment_id,
+                             std::vector<product_with_name> const& products)
   {
-    if (batch.empty())
+
+    if (products.empty())
       return;
 
-    // Look up creator from config based on product name. O(1) lookup instead of loop
-    auto it = m_product_to_config.find(batch[0].label);
+    auto it = m_product_to_config.find(products[0].label);
     if (it == m_product_to_config.end()) {
-      throw std::runtime_error("No configuration found for product: " + batch[0].label);
+      throw std::runtime_error("No configuration found for product: " + products[0].label);
     }
 
     // FIXME: Really only needed on first call
-    std::map<std::string, std::string> products;
-    for (auto const& pb : batch) {
+    std::map<std::string, std::string> product_types;
+    for (auto const& pb : products) {
       std::string const& type = m_type_map->names[pb.type];
-      products.insert(std::make_pair(pb.label, type));
+      product_types.insert(std::make_pair(pb.label, type));
     }
-    m_pers->createContainers(creator, products);
-    for (auto const& pb : batch) {
+
+    m_pers->createContainers(creator, product_types);
+
+    for (auto const& pb : products) {
       std::string const& type = m_type_map->names[pb.type];
       // FIXME: We could consider checking id to be identical for all product bases here
       m_pers->registerWrite(creator, pb.label, pb.data, type);
     }
-    // Single commit per segment (product ID shared among products in the same segment)
-    std::string const& id = batch[0].id;
-    m_pers->commitOutput(creator, id);
+
+    m_pers->commitOutput(creator, segment_id);
   }
 
-  void form_interface::read(std::string const& creator, mock_phlex::product_base& pb)
+  void form_interface::read(std::string const& creator,
+                            std::string const& segment_id,
+                            product_with_name& pb)
   {
-    // Look up creator from config based on product name. O(1) lookup instead of loop
+
     auto it = m_product_to_config.find(pb.label);
     if (it == m_product_to_config.end()) {
       throw std::runtime_error("No configuration found for product: " + pb.label);
     }
 
-    // Original type lookup
     std::string type = m_type_map->names[pb.type];
 
-    // Use full_label instead of pb.label
-    m_pers->read(creator, pb.label, pb.id, &pb.data, type);
+    m_pers->read(creator, pb.label, segment_id, &pb.data, type);
   }
 }

form/form/form.hpp

@@ -4,29 +4,49 @@
 #define __FORM_HPP__
 
 #include "form/config.hpp"
-#include "mock_phlex/phlex_toy_config.hpp"
-#include "mock_phlex/phlex_toy_core.hpp" // FORM Interface may include core phlex modules
 #include "persistence/ipersistence.hpp"
 
+#include <map>
 #include <memory>
 #include <string>
+#include <typeindex>
+#include <unordered_map>
+#include <vector>
 
 namespace form::experimental {
+
+  struct product_type_names {
+    std::unordered_map<std::type_index, std::string> names;
+  };
+
+  struct product_with_name {
+    std::string label;
+    void const* data;
+    std::type_index type;
+  };
+
   class form_interface {
   public:
-    form_interface(std::shared_ptr<mock_phlex::product_type_names> tm,
-                   mock_phlex::config::parse_config const& config);
+    form_interface(std::shared_ptr<product_type_names> tm,
+                   config::output_item_config const& output_config,
+                   config::tech_setting_config const& tech_config);
     ~form_interface() = default;
 
-    void write(std::string const& creator, mock_phlex::product_base const& pb);
     void write(std::string const& creator,
-               std::vector<mock_phlex::product_base> const& batch); // batch version
-    void read(std::string const& creator, mock_phlex::product_base& pb);
+               std::string const& segment_id,
+               product_with_name const& product);
+
+    void write(std::string const& creator,
+               std::string const& segment_id,
+               std::vector<product_with_name> const& products);
+
+    void read(std::string const& creator,
+              std::string const& segment_id,
+              product_with_name& product);
 
   private:
     std::unique_ptr<form::detail::experimental::IPersistence> m_pers;
-    std::shared_ptr<mock_phlex::product_type_names> m_type_map;
-    // Fast lookup maps built once in constructor
+    std::shared_ptr<product_type_names> m_type_map;
     std::map<std::string, form::experimental::config::PersistenceItem> m_product_to_config;
   };
 }

form/form_module.cpp

@@ -0,0 +1,156 @@
+#include "phlex/model/product_store.hpp"
+#include "phlex/model/products.hpp"
+#include "phlex/module.hpp"
+
+// FORM headers - these need to be available via CMake configuration
+// need to set up the build system to find these headers
+#include "form/config.hpp"
+#include "form/form.hpp"
+#include "form/technology.hpp"
+
+#include <iostream>
+
+namespace {
+
+  class FormOutputModule {
+  public:
+    FormOutputModule(std::shared_ptr<form::experimental::product_type_names> type_map,
+                     std::string output_file,
+                     int technology,
+                     std::vector<std::string> const& products_to_save) :
+      m_type_map(type_map), m_output_file(std::move(output_file)), m_technology(technology)
+    {
+      std::cout << "FormOutputModule initialized\n";
+      std::cout << "  Output file: " << m_output_file << "\n";
+      std::cout << "  Technology: " << m_technology << "\n";
+
+      // Build FORM configuration
+      form::experimental::config::output_item_config output_cfg;
+      form::experimental::config::tech_setting_config tech_cfg;
+
+      // FIXME: Temporary solution to accommodate Phlex limitation.
+      // Eventually, Phlex will communicate to FORM which products will be written
+      // before executing any algorithms
+
+      // Temp. Sol for Phlex Prototype 0.1
+      // Register products from config
+      for (auto const& product : products_to_save) {
+        output_cfg.addItem(product, m_output_file, m_technology);
+      }
+
+      // Initialize FORM interface
+      m_form_interface =
+        std::make_unique<form::experimental::form_interface>(type_map, output_cfg, tech_cfg);
+    }
+
+    // This method is called by Phlex - signature must be: void(product_store const&)
+    void save_data_products(phlex::experimental::product_store const& store)
+    {
+      // Check if store is empty - smart way, check store not products vector
+      if (store.empty()) {
+        return;
+      }
+
+      // STEP 1: Extract metadata from Phlex's product_store
+
+      // Extract creator (algorithm name)
+      std::string creator = store.source();
+
+      // Extract segment ID (partition) - extract once for entire store
+      std::string segment_id = store.id()->to_string();
+
+      std::cout << "\n=== FormOutputModule::save_data_products ===\n";
+      std::cout << "Creator: " << creator << "\n";
+      std::cout << "Segment ID: " << segment_id << "\n";
+      std::cout << "Number of products: " << store.size() << "\n";
+
+      // STEP 2: Convert each Phlex product to FORM format
+
+      // Collect all products for writing
+      std::vector<form::experimental::product_with_name> products;
+
+      // Reserve space for efficiency - avoid reallocations
+      products.reserve(store.size());
+
+      // Iterate through all products in the store
+      for (auto const& [product_name, product_ptr] : store) {
+        // product_name: "tracks" (from the map key)
+        // product_ptr: pointer to the actual product data
+
+        std::cout << "  Product: " << product_name << "\n";
+
+        // Create FORM product with metadata
+        products.emplace_back(product_name,           // label, from map key
+                              product_ptr->address(), // data,  from phlex product_base
+                              product_ptr->type()     // type,  from phlex product_base
+        );
+      }
+
+      // STEP 3: Send everything to FORM for persistence
+
+      // Write all products to FORM
+      // Pass segment_id once for entire collection (not duplicated in each product)
+      // No need to check if products is empty - already checked store.empty() above
+      m_form_interface->write(creator, segment_id, products);
+      std::cout << "Wrote " << products.size() << " products to FORM\n";
+    }
+
+  private:
+    std::shared_ptr<form::experimental::product_type_names> m_type_map;
+    std::string m_output_file;
+    int m_technology;
+    std::unique_ptr<form::experimental::form_interface> m_form_interface;
+  };
+
+}
+
+PHLEX_EXPERIMENTAL_REGISTER_ALGORITHMS(m, config)
+{
+  std::cout << "Registering FORM output module...\n";
+
+  // Create type map
+  auto type_map = std::make_shared<form::experimental::product_type_names>();
+
+  // Register some fundamental type for simple products
+  type_map->names[std::type_index(typeid(int))] = "int";
+  type_map->names[std::type_index(typeid(long))] = "long";
+  type_map->names[std::type_index(typeid(float))] = "float";
+  type_map->names[std::type_index(typeid(double))] = "double";
+  type_map->names[std::type_index(typeid(std::vector<int>))] = "std::vector<int>";
+  type_map->names[std::type_index(typeid(std::vector<long>))] = "std::vector<long>";
+  type_map->names[std::type_index(typeid(std::vector<float>))] = "std::vector<float>";
+  type_map->names[std::type_index(typeid(std::vector<double>))] = "std::vector<double>";
+
+  // Extract configuration from Phlex config
+  std::string output_file = config.get<std::string>("output_file", "output.root");
+  std::string tech_string = config.get<std::string>("technology", "ROOT_TTREE");
+
+  std::cout << "Configuration:\n";
+  std::cout << "  output_file: " << output_file << "\n";
+  std::cout << "  technology: " << tech_string << "\n";
+
+  // Map Phlex config string to FORM technology constant
+  int technology = form::technology::ROOT_TTREE; // default
+
+  if (tech_string == "ROOT_TTREE") {
+    technology = form::technology::ROOT_TTREE;
+  } else if (tech_string == "ROOT_RNTUPLE") {
+    technology = form::technology::ROOT_RNTUPLE;
+  } else if (tech_string == "HDF5") {
+    technology = form::technology::HDF5;
+  } else {
+    throw std::runtime_error("Unknown technology: " + tech_string);
+  }
+
+  auto products_to_save = config.get<std::vector<std::string>>("products");
+
+  // Phlex needs an OBJECT
+  // Create the FORM output module
+  auto form_output = m.make<FormOutputModule>(type_map, output_file, technology, products_to_save);
+
+  // Phlex needs a MEMBER FUNCTION to call
+  // Register the callback that Phlex will invoke
+  form_output.output("save_data_products", &FormOutputModule::save_data_products);
+
+  std::cout << "FORM output module registered successfully\n";
+}