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; }