From 062623c5ef1a0b9df0c311007154047b527c178e Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 31 Aug 2026 16:41:48 +0200 Subject: [PATCH] Analysis CCDB: add ability to have a uniformity column in a CCDB table Very often we know that CCDB objects do not change within the same run or the same dataframe (or whatever interval you can think about). For this reason it does not make sense to query for their updates for each timestamp. By specifying a "uniformity" column (e.g. RunNumber) you can short-circuit how often a given CCDB object needs to be fetched. This will eventually evolve in the ability to only have rows for unique CCDB objects fetched. --- .../CCDBSupport/src/AnalysisCCDBHelpers.cxx | 85 ++++++++++++++- Framework/Core/include/Framework/ASoA.h | 101 +++++++++++------- .../Core/include/Framework/AnalysisHelpers.h | 5 + 3 files changed, 147 insertions(+), 44 deletions(-) diff --git a/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx b/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx index fe58ea1d3746b..ecdb84072b2ae 100644 --- a/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx +++ b/Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx @@ -104,6 +104,10 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) schemaMetadata->Append("sourceMatcher", DataSpecUtils::describe(std::get(DataSpecUtils::fromMetadataString(m.defaultValue.get()).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; } @@ -144,15 +148,44 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) auto& schema = schemas[i]; std::vector ops; auto inputBinding = *schema->metadata()->Get("sourceTable"); - auto inputMatcher = DataSpecUtils::fromString(*schema->metadata()->Get("sourceMatcher")); 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()); - auto table = inputs.get(inputMatcher)->asArrowTable(); - // FIXME: make the fTimestamp column configurable. - auto timestampColumn = table->GetColumnByName("fTimestamp"); + // 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 timestampColumn; + std::shared_ptr 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(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()); @@ -179,11 +212,50 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) std::vector 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 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(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(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 responses; + for (auto ci = 0; ci < timestampColumn->num_chunks(); ++ci) { std::shared_ptr chunk = timestampColumn->chunk(ci); auto const* timestamps = chunk->data()->GetValuesSafe(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]; for (auto& field : schema->fields()) { @@ -198,7 +270,10 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/) .queryRate = 0, }); } - auto responses = CCDBFetcherHelper::populateCacheWith(helper, ops, timingInfo, dtc, allocator); + 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()); diff --git a/Framework/Core/include/Framework/ASoA.h b/Framework/Core/include/Framework/ASoA.h index 046ac57c9bf6f..2d880648c42f4 100644 --- a/Framework/Core/include/Framework/ASoA.h +++ b/Framework/Core/include/Framework/ASoA.h @@ -3335,49 +3335,72 @@ consteval auto getIndexTargets() // // The columns of this table have to be CCDB_COLUMNS so that for each timestamp, we get a row // which points to the specified CCDB objectes described by those columns. -#define DECLARE_SOA_TIMESTAMPED_TABLE_FULL(_Name_, _Label_, _TimestampSource_, _TimestampColumn_, _Version_, _Desc_, ...) \ - O2HASH(_Desc_ "/" #_Version_); \ - template \ - using _Name_##TimestampFrom = soa::Table, o2::aod::Hash<_Desc_ "/" #_Version_ ""_h>, O>; \ - using _Name_##Timestamp = _Name_##TimestampFrom>; \ - struct _Name_##TimestampMetadata : TableMetadata, __VA_ARGS__> { \ - template > \ - using base_table_t = _TimestampSource_##From; \ - template > \ - using extension_table_t = _Name_##TimestampFrom; \ - static constexpr const auto ccdb_urls = [](framework::pack) { \ - return std::array{Cs::query...}; \ - }(framework::pack<__VA_ARGS__>{}); \ - static constexpr const auto ccdb_bindings = [](framework::pack) { \ - return std::array{Cs::mLabel...}; \ - }(framework::pack<__VA_ARGS__>{}); \ - static constexpr auto N = _TimestampSource_::originals.size(); \ - template > \ - static consteval auto generateSources() \ - { \ - return _TimestampSource_##From::originals; \ - } \ - static constexpr auto timestamp_column_label = _TimestampColumn_::mLabel; \ - /*static constexpr auto timestampColumn = _TimestampColumn_;*/ \ - }; \ - template <> \ - struct MetadataTrait> { \ - static constexpr void isMetadataTrait() {}; \ - using metadata = _Name_##TimestampMetadata; \ - }; \ - template \ - using _Name_##From = o2::soa::Join<_TimestampSource_, _Name_##TimestampFrom>; \ - using _Name_ = _Name_##From \ + using _Name_##TimestampFrom = soa::Table, o2::aod::Hash<_Desc_ "/" #_Version_ ""_h>, O>; \ + using _Name_##Timestamp = _Name_##TimestampFrom>; \ + struct _Name_##TimestampMetadata : TableMetadata, __VA_ARGS__> { \ + template > \ + using base_table_t = _TimestampSource_##From; \ + template > \ + using extension_table_t = _Name_##TimestampFrom; \ + static constexpr const auto ccdb_urls = [](framework::pack) { \ + return std::array{Cs::query...}; \ + }(framework::pack<__VA_ARGS__>{}); \ + static constexpr const auto ccdb_bindings = [](framework::pack) { \ + return std::array{Cs::mLabel...}; \ + }(framework::pack<__VA_ARGS__>{}); \ + /* The uniformity column may live in a table other than the timestamp source (the run */ \ + /* number is on aod::BCs, the timestamp on aod::Timestamps). Both are handed to the */ \ + /* fetcher, which reads them positionally — sound because the two are row-aligned. */ \ + /* Row alignment cannot be checked here: ASoA encodes no type-level relation between */ \ + /* two tables that happen to have equal row counts (aod::BCs and aod::Timestamps have */ \ + /* disjoint originals). The CCDB fetcher verifies the lengths match before reading. */ \ + static constexpr auto N = o2::soa::mergeOriginals<_TimestampSource_, _UniformitySource_>().size(); \ + template > \ + static consteval auto generateSources() \ + { \ + return o2::soa::mergeOriginals<_TimestampSource_##From, _UniformitySource_##From>(); \ + } \ + static constexpr auto timestamp_column_label = _TimestampColumn_::mLabel; \ + /* Rows sharing a uniformity value resolve to the same CCDB object, so the fetcher */ \ + /* need only query once per distinct value. Defaults to the timestamp column, i.e. */ \ + /* every distinct timestamp may yield a different object — the pre-existing behaviour.*/ \ + static constexpr auto uniformity_column_label = _UniformityColumn_::mLabel; \ + /*static constexpr auto timestampColumn = _TimestampColumn_;*/ \ + }; \ + template <> \ + struct MetadataTrait> { \ + static constexpr void isMetadataTrait() {}; \ + using metadata = _Name_##TimestampMetadata; \ + }; \ + template \ + using _Name_##From = o2::soa::Join<_TimestampSource_, _Name_##TimestampFrom>; \ + using _Name_ = _Name_##From>; +/* Uniformity defaults to the timestamp column of the timestamp source: each distinct + timestamp may resolve to a different object, which is the pre-existing behaviour. + Pass an explicit uniformity source + column (e.g. aod::BCs / aod::bc::RunNumber) when + the object is constant across a coarser key: the fetcher then queries once per distinct + value instead of once per row. The uniformity source must be row-aligned with the + timestamp source, which is checked. */ #define DECLARE_SOA_TIMESTAMPED_TABLE(_Name_, _TimestampSource_, _TimestampColumn_, _Version_, _Desc_, ...) \ O2HASH(#_Name_ "Timestamped"); \ - DECLARE_SOA_TIMESTAMPED_TABLE_FULL(_Name_, #_Name_ "Timestamped", _TimestampSource_, _TimestampColumn_, _Version_, _Desc_, __VA_ARGS__) + DECLARE_SOA_TIMESTAMPED_TABLE_FULL(_Name_, #_Name_ "Timestamped", _TimestampSource_, _TimestampColumn_, _TimestampSource_, _TimestampColumn_, _Version_, _Desc_, __VA_ARGS__) + +/* Short form for a table with a coarser uniformity key; unlike the CCDB column macros the + short form is worth keeping, because going through _FULL would also make every caller + hand-write the O2HASH of the label. */ +#define DECLARE_SOA_UNIFORM_TABLE(_Name_, _TimestampSource_, _TimestampColumn_, _UniformitySource_, _UniformityColumn_, _Version_, _Desc_, ...) \ + O2HASH(#_Name_ "Timestamped"); \ + DECLARE_SOA_TIMESTAMPED_TABLE_FULL(_Name_, #_Name_ "Timestamped", _TimestampSource_, _TimestampColumn_, _UniformitySource_, _UniformityColumn_, _Version_, _Desc_, __VA_ARGS__) namespace o2::soa { diff --git a/Framework/Core/include/Framework/AnalysisHelpers.h b/Framework/Core/include/Framework/AnalysisHelpers.h index a6765e74ac637..6e046e4de0311 100644 --- a/Framework/Core/include/Framework/AnalysisHelpers.h +++ b/Framework/Core/include/Framework/AnalysisHelpers.h @@ -371,6 +371,11 @@ constexpr auto getCCDBMetadata() -> std::vector std::sort(results.begin(), results.end(), [](framework::ConfigParamSpec const& a, framework::ConfigParamSpec const& b) { return a.name < b.name; }); auto last = std::unique(results.begin(), results.end(), [](framework::ConfigParamSpec const& a, framework::ConfigParamSpec const& b) { return a.name == b.name; }); results.erase(last, results.end()); + // Tell the fetcher which column carries the timestamp to query at, and which column + // it may group by (rows sharing a uniformity value resolve to the same object, so one + // query per distinct value suffices). Both default to the timestamp column. + results.push_back({std::string{"timestamp-column"}, framework::VariantType::String, std::string{T::timestamp_column_label}, {"\"\""}}); + results.push_back({std::string{"uniformity-column"}, framework::VariantType::String, std::string{T::uniformity_column_label}, {"\"\""}}); return results; }