AnalysisCCDBHelpers.cxx

Go to the documentation of this file.

// Copyright 2019-2025 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
 
#include "AnalysisCCDBHelpers.h"
#include "CCDBFetcherHelper.h"
#include "Framework/ArrowTypes.h"
#include "Framework/DataProcessingStats.h"
#include "Framework/DeviceSpec.h"
#include "Framework/TimingInfo.h"
#include "Framework/ConfigParamRegistry.h"
#include "Framework/DataTakingContext.h"
#include "Framework/RawDeviceService.h"
#include "Framework/Output.h"
#include "Framework/Signpost.h"
#include "Framework/DanglingEdgesContext.h"
#include "Framework/ConfigContext.h"
#include "Framework/ConfigParamsHelper.h"
#include <fairmq/Version.h>
#include <arrow/array/builder_binary.h>
#include <arrow/type.h>
#include <arrow/type_fwd.h>
#include <arrow/util/key_value_metadata.h>
#include <arrow/table.h>
#include <arrow/array.h>
#include <arrow/builder.h>
#include <fmt/base.h>
#include <ctime>
#include <memory>
#include "CCDBPathTable.h"
 
#include <string>
#include <unordered_map>
#include <vector>
 
O2_DECLARE_DYNAMIC_LOG(ccdb);
 
namespace o2::framework
{
// Fill valid routes. Notice that for analysis the timestamps are associated to
// a ATIM table and there might be multiple CCDB objects of the same kind for
// dataframe.
// For this reason rather than matching the Lifetime::Condition, we match the
// origin.
namespace
{
void fillValidRoutes(CCDBFetcherHelper& helper, std::vector<o2::framework::OutputRoute> const& outputRoutes, std::unordered_map<std::string, int>& bindings)
{
  for (auto& route : outputRoutes) {
    if (std::ranges::none_of(route.matcher.metadata, [](auto const& m) { return m.name.starts_with("ccdb:"); })) {
      continue;
    }
    auto specStr = DataSpecUtils::describe(route.matcher);
    if (bindings.find(specStr) != bindings.end()) {
      continue;
    }
    bindings[specStr] = helper.routes.size();
    helper.routes.push_back(route);
    LOGP(info, "The following route needs condition objects {} ", DataSpecUtils::describe(route.matcher));
    for (auto& metadata : route.matcher.metadata) {
      if (metadata.type == VariantType::String) {
        LOGP(info, "- {}: {}", metadata.name, metadata.defaultValue.asString());
      }
    }
  }
}
} // namespace
 
AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
{
  return adaptStateful([](ConfigParamRegistry const& options, DeviceSpec const& spec, InitContext& ic) {
    auto& dec = ic.services().get<DanglingEdgesContext>();
    // The effective default for each ccdb: option was already resolved at topology
    // time by ArrowSupport (consulting task Configurables) and registered on this
    // device's options. Here we just read the final value — honouring any further
    // runtime override supplied via CLI or JSON config.
    std::unordered_map<std::string, std::string> ccdbUrls;
    std::unordered_map<std::string, std::string> runDependent;
    for (auto& input : dec.analysisCCDBInputs) {
      for (auto& m : input.metadata) {
        if (m.name.starts_with("ccdb-run-dependent:")) {
          runDependent.emplace(m.name, m.defaultValue.asString());
          continue;
        }
        if (!m.name.starts_with("ccdb:") || ccdbUrls.count(m.name)) {
          continue;
        }
        std::string url = m.defaultValue.asString();
        if (ConfigParamsHelper::hasOption(spec.options, m.name)) {
          url = options.get<std::string>(m.name.c_str());
        }
        LOGP(info, "CCDB path resolved for {}: {}", m.name, url);
        ccdbUrls.emplace(m.name, std::move(url));
      }
    }
    std::vector<std::shared_ptr<arrow::Schema>> schemas;
    for (auto& input : dec.analysisCCDBInputs) {
      auto schemaMetadata = std::make_shared<arrow::KeyValueMetadata>();
      std::vector<std::shared_ptr<arrow::Field>> fields;
      schemaMetadata->Append("outputRoute", DataSpecUtils::describe(input));
      schemaMetadata->Append("outputBinding", input.binding);
      for (auto& m : input.metadata) {
        if (m.name.starts_with("input:")) {
          auto name = m.name.substr(6);
          schemaMetadata->Append("sourceTable", name);
          schemaMetadata->Append("sourceMatcher", DataSpecUtils::describe(std::get<ConcreteDataMatcher>(DataSpecUtils::fromMetadataString(m.defaultValue.get<std::string>()).matcher)));
          continue;
        }
        if (m.name == "timestamp-column" || m.name == "uniformity-column") {
          schemaMetadata->Append(m.name, m.defaultValue.asString());
          continue;
        }
        if (!m.name.starts_with("ccdb:")) {
          continue;
        }
        auto fieldMetadata = std::make_shared<arrow::KeyValueMetadata>();
        auto it = ccdbUrls.find(m.name);
        fieldMetadata->Append("url", it != ccdbUrls.end() ? it->second : m.defaultValue.asString());
        auto runDep = runDependent.find("ccdb-run-dependent:" + m.name.substr(strlen("ccdb:")));
        fieldMetadata->Append("runDependent", runDep != runDependent.end() ? runDep->second : "0");
        auto pos = m.name.rfind(std::string_view{"::"});
        auto columnName = fmt::format("f{}", m.name.substr(pos + 2, m.name.size()));
        fields.emplace_back(std::make_shared<arrow::Field>(columnName, soa::asArrowDataType<int64_t[3]>(), false, fieldMetadata));
      }
      schemas.emplace_back(std::make_shared<arrow::Schema>(fields, schemaMetadata));
    }
 
    // Parse the declared path mappings once; they are fixed for the run of the workflow.
    std::vector<std::vector<PathTable>> pathTables;
    for (auto const& schema : schemas) {
      auto& tables = pathTables.emplace_back();
      for (auto const& field : schema->fields()) {
        tables.push_back(PathTable::parse(*field->metadata()->Get("url")));
      }
    }
 
    std::vector<std::pair<uint32_t, std::shared_ptr<arrow::FixedSizeListBuilder>>> allbuilders;
    allbuilders.resize([&schemas]() { size_t size = 0; for (auto& schema : schemas) { size += schema->num_fields(); }; return size; }());
    auto* pool = arrow::default_memory_pool();
 
    int idx = 0;
    int sidx = 0;
    for (auto const& schema : schemas) {
      for (auto const& _ : schema->fields()) {
        auto value_builder = std::make_shared<arrow::Int64Builder>();
        allbuilders[idx] = std::make_pair(sidx, std::make_shared<arrow::FixedSizeListBuilder>(pool, std::move(value_builder), 3));
        ++idx;
      }
      ++sidx;
    }
 
    std::shared_ptr<CCDBFetcherHelper> helper = std::make_shared<CCDBFetcherHelper>();
    CCDBFetcherHelper::initialiseHelper(*helper, options);
    std::unordered_map<std::string, int> bindings;
    fillValidRoutes(*helper, spec.outputs, bindings);
 
    return adaptStateless([schemas, bindings, helper, allbuilders, pathTables](InputRecord& inputs, DataTakingContext& dtc, DataAllocator& allocator, TimingInfo& timingInfo, DataProcessingStats& stats) {
      O2_SIGNPOST_ID_GENERATE(sid, ccdb);
      O2_SIGNPOST_START(ccdb, sid, "fetchFromAnalysisCCDB", "Fetching CCDB objects for analysis%" PRIu64, (uint64_t)timingInfo.timeslice);
      std::ranges::for_each(allbuilders, [](auto& builder) { builder.second->Reset(); });
      for (auto i = 0U; i < schemas.size(); ++i) {
        auto& schema = schemas[i];
        std::vector<CCDBFetcherHelper::FetchOp> ops;
        auto inputBinding = *schema->metadata()->Get("sourceTable");
        auto outRouteDesc = *schema->metadata()->Get("outputRoute");
        std::string outBinding = *schema->metadata()->Get("outputBinding");
        auto timestampColumnName = schema->metadata()->Contains("timestamp-column") ? *schema->metadata()->Get("timestamp-column") : std::string{"fTimestamp"};
        auto uniformityColumnName = schema->metadata()->Contains("uniformity-column") ? *schema->metadata()->Get("uniformity-column") : timestampColumnName;
        O2_SIGNPOST_EVENT_EMIT_INFO(ccdb, sid, "fetchFromAnalysisCCDB",
                                    "Fetching CCDB objects for %{public}s's columns with timestamps from %{public}s and putting them in route %{public}s",
                                    outBinding.c_str(), inputBinding.c_str(), outRouteDesc.c_str());
        // The timestamp and uniformity columns may live in different source tables (the
        // run number is on aod::BCs, the timestamp on aod::Timestamps). Locate each by
        // name across every declared source, and read them positionally.
        std::shared_ptr<arrow::ChunkedArray> timestampColumn;
        std::shared_ptr<arrow::ChunkedArray> uniformityColumn;
        auto const& schemaKeys = schema->metadata()->keys();
        auto const& schemaValues = schema->metadata()->values();
        for (size_t mi = 0; mi < schemaKeys.size(); ++mi) {
          if (schemaKeys[mi] != "sourceMatcher") {
            continue;
          }
          auto sourceTable = inputs.get<TableConsumer>(DataSpecUtils::fromString(schemaValues[mi]))->asArrowTable();
          if (auto column = sourceTable->GetColumnByName(timestampColumnName); column && !timestampColumn) {
            timestampColumn = column;
          }
          if (auto column = sourceTable->GetColumnByName(uniformityColumnName); column && !uniformityColumn) {
            uniformityColumn = column;
          }
        }
        if (!timestampColumn) {
          LOGP(fatal, "No source table of {} provides the timestamp column \"{}\"", outBinding, timestampColumnName);
        }
        if (!uniformityColumn) {
          LOGP(fatal, "No source table of {} provides the uniformity column \"{}\"", outBinding, uniformityColumnName);
        }
        // Positional reading is only sound if the two sources are row-aligned; ASoA has
        // no type-level way to state that, so it is checked here.
        if (uniformityColumn->length() != timestampColumn->length()) {
          LOGP(fatal, "Uniformity column \"{}\" has {} rows but timestamp column \"{}\" has {}; the two sources of {} are not row-aligned",
               uniformityColumnName, uniformityColumn->length(), timestampColumnName, timestampColumn->length(), outBinding);
        }
        auto reserveSize = timestampColumn->length();
        O2_SIGNPOST_EVENT_EMIT_INFO(ccdb, sid, "fetchFromAnalysisCCDB",
                                    "There are %zu bindings available", bindings.size());
        for (auto const& binding : bindings) {
          O2_SIGNPOST_EVENT_EMIT_INFO(ccdb, sid, "fetchFromAnalysisCCDB",
                                      "* %{public}s: %d",
                                      binding.first.c_str(), binding.second);
        }
        int outputRouteIndex = bindings.at(outRouteDesc);
        auto& spec = helper->routes[outputRouteIndex].matcher;
        auto concrete = DataSpecUtils::asConcreteDataMatcher(spec);
        Output output{concrete.origin, concrete.description, concrete.subSpec};
        auto builders = allbuilders | std::views::filter([&i](auto const& builder) { return builder.first == i; });
        unsigned int numBuilders = std::ranges::count_if(allbuilders, [&i](auto const& builder) { return builder.first == i; });
        arrow::Status status;
        std::ranges::for_each(builders, [&status, &reserveSize](auto& builder) {
          if (reserveSize > builder.second->capacity()) {
            status &= builder.second->Reserve(reserveSize - builder.second->capacity());
          }
        });
        if (!status.ok()) {
          throw framework::runtime_error_f("Failed to reserve arrays: ", status.ToString().c_str());
        }
 
        std::vector<DataAllocator::CacheId> lastIds(numBuilders, DataAllocator::CacheId{.value = -1, .handle = -1, .segment = -1});
 
        // Rows sharing a uniformity value resolve to the same objects, so the query is
        // issued once per distinct value and the resulting handles are repeated for the
        // rest of the run. When uniformity is the timestamp itself (the default) this
        // degenerates to the previous behaviour, one query per row.
        std::vector<int64_t> uniformity;
        bool const shortCircuit = uniformityColumn.get() != timestampColumn.get();
        if (shortCircuit) {
          uniformity.reserve(reserveSize);
          for (auto uci = 0; uci < uniformityColumn->num_chunks(); ++uci) {
            auto uchunk = uniformityColumn->chunk(uci);
            auto const length = uchunk->data()->length;
            switch (uchunk->type_id()) {
              case arrow::Type::INT32:
                for (int64_t ui = 0; ui < length; ++ui) {
                  uniformity.push_back(uchunk->data()->GetValuesSafe<int32_t>(1)[ui]);
                }
                break;
              case arrow::Type::INT64:
              case arrow::Type::UINT64:
                for (int64_t ui = 0; ui < length; ++ui) {
                  uniformity.push_back(uchunk->data()->GetValuesSafe<int64_t>(1)[ui]);
                }
                break;
              default:
                LOGP(fatal, "Uniformity column \"{}\" of {} has unsupported arrow type {}",
                     uniformityColumnName, outBinding, uchunk->type()->ToString());
            }
          }
        }
        int64_t row = -1;
        int64_t previousUniformity = 0;
        bool haveResponses = false;
        std::vector<CCDBFetcherHelper::Response> responses;
 
        for (auto ci = 0; ci < timestampColumn->num_chunks(); ++ci) {
          std::shared_ptr<arrow::Array> chunk = timestampColumn->chunk(ci);
          auto const* timestamps = chunk->data()->GetValuesSafe<size_t>(1);
 
          for (int64_t ri = 0; ri < chunk->data()->length; ri++) {
            ++row;
            bool const sameAsPrevious = shortCircuit && haveResponses && uniformity[row] == previousUniformity;
            if (shortCircuit) {
              previousUniformity = uniformity[row];
            }
            ops.clear();
            int64_t timestamp = timestamps[ri];
            // Key the path lookup on the uniformity value; when uniformity is the
            // timestamp itself the mapping expresses validity intervals instead.
            int64_t const uniformityKey = shortCircuit ? uniformity[row] : timestamp;
            int fi = 0;
            for (auto& field : schema->fields()) {
              auto const& url = pathTables[i][fi++].resolve(uniformityKey, field->name());
              // Time to actually populate the blob
              // A run-dependent object is queried with the run number rather than by
              // timestamp alone. The run comes from the uniformity value, so the column's
              // table has to be uniform in the run number for this to mean anything.
              int const fieldRunDependent = field->metadata()->Contains("runDependent")
                                              ? std::stoi(*field->metadata()->Get("runDependent"))
                                              : 0;
              if (fieldRunDependent != 0 && uniformityColumnName != "fRunNumber") {
                LOGP(fatal, R"(Column "{}" of {} is declared run-dependent, but its table is uniform in "{}" rather than fRunNumber, so no run number is available to query with. Declare the table with DECLARE_SOA_UNIFORM_TABLE(..., aod::BCs, o2::aod::bc::RunNumber, ...).)",
                     field->name(), outBinding, uniformityColumnName);
              }
              ops.push_back({
                .spec = spec,
                .url = url,
                .timestamp = timestamp,
                .runNumber = fieldRunDependent != 0 ? static_cast<int>(uniformityKey) : 1,
                .runDependent = fieldRunDependent,
                .queryRate = 0,
              });
            }
            if (!sameAsPrevious) {
              responses = CCDBFetcherHelper::populateCacheWith(helper, ops, timingInfo, dtc, allocator);
              haveResponses = true;
            }
            O2_SIGNPOST_START(ccdb, sid, "handlingResponses",
                              "Got %zu responses from server.",
                              responses.size());
            if (numBuilders != responses.size()) {
              LOGP(fatal, "Not enough responses (expected {}, found {})", numBuilders, responses.size());
            }
            arrow::Status result;
 
            int bi = 0;
            for (auto& builder : builders) {
              auto& response = responses[bi];
              auto& lastId = lastIds[bi];
              if (response.id.value != lastId.value) {
                lastId.value = response.id.value;
                allocator.adoptFromCache(output, response.id, header::gSerializationMethodCCDB);
              }
              result &= builder.second->Append();
              auto* value_builder = dynamic_cast<arrow::Int64Builder*>(builder.second->value_builder());
              result &= value_builder->Append(response.id.handle);
              result &= value_builder->Append(response.id.segment);
              result &= value_builder->Append(response.size);
              ++bi;
            }
            if (!result.ok()) {
              LOGP(fatal, "Error adding results from CCDB");
            }
            O2_SIGNPOST_END(ccdb, sid, "handlingResponses", "Done processing responses");
          }
        }
        arrow::ArrayVector arrays;
        std::ranges::for_each(builders, [&arrays](auto& builder) { arrays.push_back(*builder.second->Finish()); });
        auto outTable = arrow::Table::Make(schema, arrays);
        allocator.adopt(output, outTable);
      }
 
      stats.updateStats({(int)ProcessingStatsId::CCDB_CACHE_FETCHED_BYTES, DataProcessingStats::Op::Set, (int64_t)helper->totalFetchedBytes});
      stats.updateStats({(int)ProcessingStatsId::CCDB_CACHE_REQUESTED_BYTES, DataProcessingStats::Op::Set, (int64_t)helper->totalRequestedBytes});
      O2_SIGNPOST_END(ccdb, sid, "fetchFromAnalysisCCDB", "Fetching CCDB objects");
    });
  });
}
 
} // namespace o2::framework