Prepare Run3_2024, Run3_2025 and Run3_2026 (shared MC + weight_base_cmb)

.github/workflows/cross-section-check.yaml

@@ -47,4 +47,4 @@ jobs:
 
       - name: Check cross-sections
         if: ${{ steps.changed_files.outputs.has_changes == 'true' }}
-        run: python3 test/checkCrossSections.py Run3_2022 Run3_2022EE Run3_2023 Run3_2023BPix Run3_2024 Run3_2025
+        run: python3 test/checkCrossSections.py Run3_2022 Run3_2022EE Run3_2023 Run3_2023BPix Run3_2024 Run3_2025 Run3_2026

.github/workflows/ds-consistency-check.yaml

@@ -45,8 +45,8 @@ jobs:
 
       - name: Check dataset configs consistency
         if: ${{ steps.changed_files.outputs.has_ds_configs == 'true' }}
-        run: python3 test/checkDatasetConfigConsistency.py --exception config/dataset_exceptions.yaml Run3_2022 Run3_2022EE Run3_2023 Run3_2023BPix Run3_2024 Run3_2025
+        run: python3 test/checkDatasetConfigConsistency.py --exception config/dataset_exceptions.yaml Run3_2022 Run3_2022EE Run3_2023 Run3_2023BPix Run3_2024 Run3_2025 Run3_2026
 
       - name: Check dataset naming
         if: ${{ steps.changed_files.outputs.has_ds_configs == 'true' }}
-        run: python3 test/checkDatasetNaming.py --rules config/dataset_naming_rules.yaml Run3_2022 Run3_2022EE Run3_2023 Run3_2023BPix Run3_2024 Run3_2025
+        run: python3 test/checkDatasetNaming.py --rules config/dataset_naming_rules.yaml Run3_2022 Run3_2022EE Run3_2023 Run3_2023BPix Run3_2024 Run3_2025 Run3_2026

AnaProd/MergeAnaTuples.py

@@ -55,17 +55,13 @@
     """)
 
 
-def combineAnaCaches(anaCaches, processors):
-    """
-    Combine multiple anaCaches into one.
-    Merges denominators, runtimes, and any processor-provided sections (like DY_stitching).
-    """
-    if len(anaCaches) == 0:
-        raise RuntimeError("addAnaCaches: no anaCaches provided")
+def _combine_denominator_map(anaCaches, processors, key):
     denominator = {}
     anaCache_processors = set()
     for anaCache in anaCaches:
-        for source, source_entry in anaCache["denominator"].items():
+        if key not in anaCache:
+            raise RuntimeError(f"combineAnaCaches: cache is missing '{key}'")
+        for source, source_entry in anaCache[key].items():
             if source not in denominator:
                 denominator[source] = {}
             for scale in getScales(source):
@@ -82,24 +78,35 @@ def combineAnaCaches(anaCaches, processors):
                 entries = []
                 for anaCache in anaCaches:
                     if (
-                        source in anaCache["denominator"]
-                        and scale in anaCache["denominator"][source]
-                        and processor in anaCache["denominator"][source][scale]
+                        source in anaCache[key]
+                        and scale in anaCache[key][source]
+                        and processor in anaCache[key][source][scale]
                     ):
-                        entries.append(
-                            anaCache["denominator"][source][scale][processor]
-                        )
+                        entries.append(anaCache[key][source][scale][processor])
                     else:
                         raise RuntimeError(
-                            f"combineAnaCaches: missing entry for {source}/{scale}/{processor} in one of the caches"
+                            f"combineAnaCaches: missing entry for {key}/{source}/{scale}/{processor} in one of the caches"
                         )
                 denominator[source][scale][processor] = processors[
                     processor
                 ].onAnaCache_combineAnaCaches(entries)
+    return denominator
+
 
+def combineAnaCaches(anaCaches, processors):
+    """
+    Combine multiple anaCaches into one.
+    Merges denominators, runtimes, and any processor-provided sections (like DY_stitching).
+    """
+    if len(anaCaches) == 0:
+        raise RuntimeError("addAnaCaches: no anaCaches provided")
     anaCacheSum = {
-        "denominator": denominator,
+        "denominator": _combine_denominator_map(anaCaches, processors, "denominator"),
     }
+    if any("denominator_cmb" in anaCache for anaCache in anaCaches):
+        anaCacheSum["denominator_cmb"] = _combine_denominator_map(
+            anaCaches, processors, "denominator_cmb"
+        )
     return anaCacheSum
 
 

AnaProd/anaTupleProducer.py

@@ -16,6 +16,7 @@
 import FLAF.Common.ReportTools as ReportTools
 import FLAF.Common.triggerSel as Triggers
 from FLAF.Common.Setup import Setup
+from FLAF.Common.shared_mc import shared_mc_in_era_expr, shared_mc_split
 from Corrections.Corrections import Corrections
 from Corrections.lumi import LumiFilter
 from Corrections.CorrectionsCore import central, getScales, getSystName
@@ -180,19 +181,30 @@ def createAnatuple(
     if "pu" in corrections.to_apply and compute_unc_variations:
         shape_sources += puWeightProducer.uncSource
 
-    report["denominator"] = {}
-    for shape_unc_source in shape_sources:
-        report["denominator"][shape_unc_source] = {}
-        for shape_unc_scale in getScales(shape_unc_source):
-            report["denominator"][shape_unc_source][shape_unc_scale] = {}
-            for p_name, p_instance in processor_instances.items():
-                report["denominator"][shape_unc_source][shape_unc_scale][
-                    p_name
-                ] = p_instance.onAnaCache_initializeDenomEntry()
+    shared_mc = None if isData else setup.global_params.get("shared_mc")
+    shared_mc_expr = None
+    if shared_mc:
+        split_mod, lo, hi, _ = shared_mc_split(period, shared_mc)
+        shared_mc_expr = shared_mc_in_era_expr(split_mod, lo, hi)
+
+    def initializeDenomReport(key):
+        report[key] = {}
+        for shape_unc_source in shape_sources:
+            report[key][shape_unc_source] = {}
+            for shape_unc_scale in getScales(shape_unc_source):
+                report[key][shape_unc_source][shape_unc_scale] = {}
+                for p_name, p_instance in processor_instances.items():
+                    report[key][shape_unc_source][shape_unc_scale][
+                        p_name
+                    ] = p_instance.onAnaCache_initializeDenomEntry()
+
+    initializeDenomReport("denominator")
+    if shared_mc_expr:
+        initializeDenomReport("denominator_cmb")
 
     gen_weight_name = "weight_gen"
 
-    def updateDenomEntry(rdf):
+    def updateDenomEntry(rdf, report_key, branch_prefix):
         for p_instance in processor_instances.values():
             rdf = p_instance.onAnaCache_prepareDataFrame(rdf)
 
@@ -203,10 +215,10 @@ def updateDenomEntry(rdf):
                 if "pu" in corrections.to_apply:
                     weights_to_apply.append(f"weight_pu_{shape_unc_scale}")
                 for p_name, p_instance in processor_instances.items():
-                    output_branch_name = f"weight_denom_{p_name}_{shape_unc_name}"
-                    report["denominator"][shape_unc_source][shape_unc_scale][p_name] = (
+                    output_branch_name = f"{branch_prefix}_{p_name}_{shape_unc_name}"
+                    report[report_key][shape_unc_source][shape_unc_scale][p_name] = (
                         p_instance.onAnaCache_updateDenomEntry(
-                            report["denominator"][shape_unc_source][shape_unc_scale][
+                            report[report_key][shape_unc_source][shape_unc_scale][
                                 p_name
                             ],
                             rdf,
@@ -228,7 +240,14 @@ def updateDenomEntry(rdf):
             data_frame = data_frame.Define(gen_weight_name, genWeight_def)
             if "pu" in corrections.to_apply:
                 data_frame = corrections.pu.getWeight(data_frame)
-            updateDenomEntry(data_frame)
+            updateDenomEntry(data_frame, "denominator", "weight_denom")
+            if shared_mc_expr:
+                data_frame = data_frame.Define("__shared_mc_in_era", shared_mc_expr)
+                updateDenomEntry(
+                    data_frame.Filter("__shared_mc_in_era"),
+                    "denominator_cmb",
+                    "weight_denom_cmb",
+                )
     # if isData: json_dict_for_cache['RunLumi'] = unique_run_lumi
 
     if range is not None:
@@ -388,19 +407,23 @@ def updateDenomEntry(rdf):
 
     report["run_lumi_ranges"] = runLumiRanges
 
-    for shape_unc_source in shape_sources:
-        for shape_unc_scale in getScales(shape_unc_source):
-            for p_name, p_instance in processor_instances.items():
-                report["denominator"][shape_unc_source][shape_unc_scale][p_name] = (
-                    p_instance.onAnaCache_materializeDenomEntry(
-                        report["denominator"][shape_unc_source][shape_unc_scale][p_name]
+    denom_keys = ["denominator"]
+    if "denominator_cmb" in report:
+        denom_keys.append("denominator_cmb")
+    for denom_key in denom_keys:
+        for shape_unc_source in shape_sources:
+            for shape_unc_scale in getScales(shape_unc_source):
+                for p_name, p_instance in processor_instances.items():
+                    report[denom_key][shape_unc_source][shape_unc_scale][p_name] = (
+                        p_instance.onAnaCache_materializeDenomEntry(
+                            report[denom_key][shape_unc_source][shape_unc_scale][p_name]
+                        )
                     )
-                )
-                report["denominator"][shape_unc_source][shape_unc_scale][p_name] = (
-                    p_instance.onAnaCache_finalizeDenomEntry(
-                        report["denominator"][shape_unc_source][shape_unc_scale][p_name]
+                    report[denom_key][shape_unc_source][shape_unc_scale][p_name] = (
+                        p_instance.onAnaCache_finalizeDenomEntry(
+                            report[denom_key][shape_unc_source][shape_unc_scale][p_name]
+                        )
                     )
-                )
 
     hist_time = ROOT.TH1D(f"time", f"time", 1, 0, 1)
     end_time = datetime.datetime.now()

Analysis/HistProducerFromNTuple.py

@@ -13,6 +13,12 @@
 from FLAF.Common.Setup import Setup
 from FLAF.RunKit.run_tools import ps_call
 from FLAF.Analysis.HistTupleProducer import DefineBinnedColumn
+from FLAF.Analysis.histFromNtupleBatch import (
+    count_booked_hists,
+    iter_hist_batches,
+    n_cut_slots,
+    unc_scale_pairs,
+)
 
 
 def find_keys(inFiles_list):
@@ -185,6 +191,14 @@ def BuildAllHistActions(
     parser.add_argument("--LAWrunVersion", required=True, type=str)
     parser.add_argument("--nMT", type=int, default=8)
     parser.add_argument("--user-custom", type=str, default=None)
+    parser.add_argument(
+        "--max-hists",
+        type=int,
+        default=None,
+        help="Max histograms booked in one RDataFrame pass, counting every "
+        "(variable, selection, unc, scale) including Up/Down. "
+        "0 disables batching. Default: hist_from_ntuple_max_hists from config, else 4000.",
+    )
     args = parser.parse_args()
 
     ROOT.EnableImplicitMT(args.nMT)
@@ -281,33 +295,62 @@ def BuildAllHistActions(
                     )
 
     if all_trees:
-        # Open a tmp ROOT file per variable, then register all histogram actions sharing one
-        # filtered RDataFrame node per selection across variables (see BuildAllHistActions).
-        # Collecting every action before triggering lets ROOT execute them in a single
-        # event-loop pass over the input files.
-        # Write each variable's histograms directly into its final, compressed output file.
-        # SaveHist persists objects via WriteTObject as the actions run, so once the single
-        # event loop has executed we just close the files -- no per-variable hadd recompress
-        # pass (209 == LZMA level 9, matching the previous `hadd -f209` output compression).
+        # Open one compressed output file per variable. SaveHist writes into it as
+        # each batch's event loop runs (209 == LZMA level 9).
         var_tmp_files = {}
         for var in vars_to_process:
             out_path = os.path.join(args.outDir, f"{var}.root")
             out_root_file = ROOT.TFile(out_path, "RECREATE", "", 209)
             var_tmp_files[var] = (out_path, out_root_file)
 
-        all_save_fns = BuildAllHistActions(
-            uncs_to_compute,
-            unc_cfg_dict,
-            all_trees,
-            vars_to_process,
-            key_filter_dict,
-            further_cuts,
-            treeName,
-            var_tmp_files,
+        if args.max_hists is not None:
+            max_hists = args.max_hists
+        else:
+            max_hists = int(setup.global_params.get("hist_from_ntuple_max_hists", 4000))
+        n_total = count_booked_hists(
+            max(1, len(vars_to_process)),
+            max(1, len(key_filter_dict)),
+            n_cut_slots(further_cuts),
+            max(1, len(unc_scale_pairs(uncs_to_compute))),
+        )
+        batches = list(
+            iter_hist_batches(
+                uncs_to_compute,
+                key_filter_dict,
+                further_cuts,
+                vars_to_process,
+                max_hists,
+            )
+        )
+        print(
+            f"Booking {n_total} histograms "
+            f"({len(vars_to_process)} vars × {len(key_filter_dict)} keys × "
+            f"{n_cut_slots(further_cuts)} cuts × "
+            f"{len(unc_scale_pairs(uncs_to_compute))} unc/scales); "
+            f"max_hists={max_hists} → {len(batches)} batch(es)"
         )
 
-        for fn in all_save_fns:
-            fn()
+        for batch_idx, (b_uncs, b_keys, b_cuts, b_vars) in enumerate(batches, start=1):
+            n_batch = count_booked_hists(
+                max(1, len(b_vars)),
+                max(1, len(b_keys)),
+                n_cut_slots(b_cuts),
+                max(1, len(unc_scale_pairs(b_uncs))),
+            )
+            print(f"Histogram batch {batch_idx}/{len(batches)}: {n_batch} hists")
+            save_fns = BuildAllHistActions(
+                b_uncs,
+                unc_cfg_dict,
+                all_trees,
+                b_vars,
+                b_keys,
+                b_cuts,
+                treeName,
+                var_tmp_files,
+            )
+            for fn in save_fns:
+                fn()
+            del save_fns
 
         for var in vars_to_process:
             _, out_root_file = var_tmp_files[var]