Repository: dmwm/wmcore
Issue: Duplicate lumi section in ACDC for MCFakeFile

@vlimant was asking me about duplicate lumi sections that are lately showing up too often in production.
This is the JIRA ticket for that prepID:
https://its.cern.ch/jira/browse/CMSCOMPPR-4903

and here is the Unified report (search for duplicate):
https://cms-unified.web.cern.ch/cms-unified/showlog/?search=task_SUS-RunIIFall18wmLHEGS-00025&module=checkor&limit=5000

Jean-Roch also provided me with the lumi number and files that have duplicates, e.g.:
```
(1, 252052) [u’/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/8D858631-45A0-4D40-A880-F5298F195329.root’, u’/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/B417C89D-5345-AA4B-BA2B-29E4F154AB6E.root’]

(1, 240646) [u’/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/78504BE6-7D0C-504D-AC33-4633FB046646.root’, u’/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/99D23191-62CC-8A4B-8DDF-175BB2786FFC.root’]
```

Investigating the lumi `240646` and making sure those 2 last files really contain the same lumi number:
https://cmsweb.cern.ch/dbs/prod/global/DBSReader/files?logical_file_name=/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/78504BE6-7D0C-504D-AC33-4633FB046646.root&run_num=1&lumi_list=[240646]
https://cmsweb.cern.ch/dbs/prod/global/DBSReader/files?logical_file_name=/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/99D23191-62CC-8A4B-8DDF-175BB2786FFC.root&run_num=1&lumi_list=[240646]

and the answer is yes, they both have lumi 240646 (second one is invalid already). They belong to the following blocks:
```
/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/RunIIAutumn18MiniAOD-102X_upgrade2018_realistic_v15-v1/MINIAODSIM#db434bf2-1827-484d-93b6-2317a2a9d422
/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/RunIIAutumn18MiniAOD-102X_upgrade2018_realistic_v15-v1/MINIAODSIM#6070c12c-9db4-4a2e-bc06-31210e06d6dc
```

Mapping these files to blocks and then investigating which agents produced those output datasets, I found that vocms0257 is where this workflow ran
vlimant_ACDC0_task_SUS-RunIIFall18wmLHEGS-00025__v1_T_190218_145226_481

Given the file names, this is the query to find which workflow produced those files:
```
SELECT dbsbuffer_block.id AS blockid, dbsbuffer_block.blockname AS blockname,
 dbsbuffer_file.id AS fileid, dbsbuffer_file.lfn AS lfn,
 dbsbuffer_workflow.id AS wfid, dbsbuffer_workflow.name AS name FROM dbsbuffer_block
 INNER JOIN dbsbuffer_file ON dbsbuffer_block.id = dbsbuffer_file.block_id
 INNER JOIN dbsbuffer_workflow ON dbsbuffer_file.workflow = dbsbuffer_workflow.id
 WHERE dbsbuffer_file.lfn
 IN ('/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/78504BE6-7D0C-504D-AC33-4633FB046646.root',
 '/store/mc/RunIIAutumn18MiniAOD/TTJets_SingleLeptFromT_TuneCP5_13TeV-madgraphMLM-pythia8/MINIAODSIM/102X_upgrade2018_realistic_v15-v1/70000/99D23191-62CC-8A4B-8DDF-175BB2786FFC.root');
``` 

After that, I wrote a script to fetch data from the ACDCServer and print files with those duplicate lumi sections:
```
cmst1@vocms0257:/data/srv/wmagent/current $ python getACDCChunkFiles.py 
Splitted block name: {'SpecName': 'vlimant_ACDC0_task_SUS-RunIIFall18wmLHEGS-00025__v1_T_190218_145226_481', 'TaskName': '/pdmvserv_task_SUS-RunIIFall18wmLHEGS-00025__v1_T_181211_005112_2222/SUS-RunIIFall18wmLHEGS-00025_0', 'NumOfFiles': 31055, 'Offset': 0}
Retrieved 156 unique files from the ACDCServer
File: MCFakeFile-pdmvserv_task_SUS-RunIIFall18wmLHEGS-00025__v1_T_181211_005112_2222-SUS-RunIIFall18wmLHEGS-00025_0-46f9bc176a3d2f7e0ecaaa75cd99cf87 with events: 528768, contains these lumis: set([240646])
File: MCFakeFile-pdmvserv_task_SUS-RunIIFall18wmLHEGS-00025__v1_T_181211_005112_2222-SUS-RunIIFall18wmLHEGS-00025_0-fcadce0757360738dec334f0c2abb6a8 with events: 566304, contains these lumis: set([252052])
Spent 77.5486171246 secs running so far
```

and as we can see, there are no dups in the ACDCServer. So something must have happened in the agent.

I also compared the job splitting between the original and the ACDC and they are the same, so there should really be no issue with that. Next steps is to investigate the jobs that produced those files (and possible the jobs that produced the respective unmerged files...)

Changed files:
src/python/WMCore/JobSplitting/EventBased.py, test/python/WMCore_t/ACDC_t/DataCollectionService_t.py, src/python/WMCore/ACDC/DataCollectionService.py, src/python/WMCore/WMBS/MySQL/Files/GetBulkRunLumi.py

diff --git a/src/python/WMCore/ACDC/DataCollectionService.py b/src/python/WMCore/ACDC/DataCollectionService.py
index 4d0a0e515..b7b8e0725 100644
--- a/src/python/WMCore/ACDC/DataCollectionService.py
+++ b/src/python/WMCore/ACDC/DataCollectionService.py
@@ -27,8 +27,10 @@ def mergeFilesInfo(chunkFiles):
     """
     _mergeFilesInfo_
 
-    Receive a list of dicts with acdc files info and merge them together.
-    Different process for different input type (data x fake file)
+    Receive a list of dicts with acdc files information and merge them when
+    it belongs to the same file. It also removes any possible duplicate lumi
+    section (for the same file) in case ErrorHandler happened to upload the
+    same job error document twice.
     """
     mergedFiles = {}
 
@@ -36,29 +38,72 @@ def mergeFilesInfo(chunkFiles):
     if chunkFiles[0]['lfn'].startswith('MCFakeFile'):
         logging.info("Merging %d ACDC FakeFiles...", len(chunkFiles))
         for acdcFile in chunkFiles:
-            if acdcFile['lfn'] not in mergedFiles:
-                mergedFiles[acdcFile['lfn']] = acdcFile
+            fName = acdcFile['lfn']
+            if fName not in mergedFiles:
+                mergedFiles[fName] = acdcFile
             else:
-                mergedFiles[acdcFile['lfn']]['events'] += acdcFile['events']
-                mergedFiles[acdcFile['lfn']]['runs'][0]['lumis'].extend(acdcFile['runs'][0]['lumis'])
-        logging.info("resulted in %d final ACDC FakeFiles.", len(mergedFiles))
+                lumiSet = set(acdcFile['runs'][0]['lumis'])
+                if lumiSet.issubset(set(mergedFiles[fName]['runs'][0]['lumis'])):
+                    # every element in lumiSet is already in mergedFiles, it's a dup!
+                    continue
+                mergedFiles[fName]['events'] += acdcFile['events']
+                mergedFiles[fName]['runs'][0]['lumis'].extend(acdcFile['runs'][0]['lumis'])
     else:
         logging.info("Merging %d real input files...", len(chunkFiles))
         for acdcFile in chunkFiles:
             fName = acdcFile['lfn']
-            if acdcFile['lfn'] not in mergedFiles:
+            if fName not in mergedFiles:
                 mergedFiles[fName] = acdcFile
             else:
+                # if one run/lumi pair is there, then it's a duplicate job
+                runNum = acdcFile['runs'][0]['run_number']
+                lumiSet = set(acdcFile['runs'][0]['lumis'])
+                if _isRunMaskDuplicate(runNum, lumiSet, mergedFiles[fName]['runs']):
+                    continue
                 # union of parents
                 allParents = list(set(mergedFiles[fName]['parents']).union(acdcFile['parents']))
                 mergedFiles[fName]['parents'] = allParents
                 # just add up run/lumi pairs (don't try to merge them)
                 mergedFiles[fName]['runs'].extend(acdcFile['runs'])
-        logging.info("resulted in %d final real files.", len(mergedFiles))
+        _mergeRealDataRunLumis(mergedFiles)
 
+    logging.info(" ... resulted in %d unique files.", len(mergedFiles))
     return mergedFiles.values()
 
 
+def _isRunMaskDuplicate(run, lumis, runLumis):
+    """
+    Test whether run and lumi is a subset of one of the
+    runLumi pairs in runLumis
+    :param run: integer run number
+    :param lumi: set of lumis
+    :param runLumis: list of dictionaries containing run and lumis
+    """
+    for runLumi in runLumis:
+        if run == runLumi['run_number']:
+            if lumis.issubset(runLumi['lumis']):
+                return True
+    return False
+
+def _mergeRealDataRunLumis(mergedFiles):
+    """
+    Function to scan and merge run/lumi pairs in the same file, thus
+    getting rid of duplicate lumi sections in real data.
+    :param mergedFiles: list of dictionaries with ACDC file info
+    :return: update data structure in place
+    """
+    for fname in mergedFiles:
+        runLumis = {}
+        for item in mergedFiles[fname]['runs']:
+            runLumis.setdefault(item['run_number'], [])
+            runLumis[item['run_number']].extend(item['lumis'])
+        # now write those back to the original data structure
+        mergedFiles[fname]['runs'] = []
+        for run, lumis in runLumis.items():
+            mergedFiles[fname]['runs'].append({'run_number': run,
+                                               'lumis': list(set(lumis))})
+    return
+
 class ACDCDCSException(WMException):
     """
     Yet another dummy variable class
@@ -305,12 +350,13 @@ class DataCollectionService(CouchService):
         Query ACDC for all of the files in the given collection and task.
         Return an entry for each file with lumis and event info.
         Format is:
-        [{'lfn' : 'someLfn',
-          'runs' : { run0 : [lumi0,lumi1], },
-          'events' :}]
+        [{'lfn': 'someLfn',
+          'first_event': XXX,
+          'lumis': [lumi0,lumi1],
+          'events': XXX}]
         """
-
         files = self._getFilesetInfo(collectionID, taskName)
+        files = mergeFilesInfo(files)
 
         acdcInfo = []
         for value in files:
diff --git a/src/python/WMCore/JobSplitting/EventBased.py b/src/python/WMCore/JobSplitting/EventBased.py
index 93f3bb843..59ec832f5 100644
--- a/src/python/WMCore/JobSplitting/EventBased.py
+++ b/src/python/WMCore/JobSplitting/EventBased.py
@@ -56,7 +56,10 @@ class EventBased(JobFactory):
                 couchDB = kwargs.get('couchDB')
                 filesetName = kwargs.get('filesetName')
                 collectionName = kwargs.get('collectionName')
-                logging.info('Creating jobs for ACDC fileset %s', filesetName)
+                logging.info('Loading ACDC info for collectionName: %s, with filesetName: %s', collectionName, filesetName)
+                ### FIXME Wait, don't we have all this data in WMBS already? Why querying
+                # the ACDCServer again to get the same data that is already stored
+                # in the WMBS table!!!!!?!?!
                 dcs = DataCollectionService(couchURL, couchDB)
                 acdcFileList = dcs.getProductionACDCInfo(collectionName, filesetName)
             except Exception as ex:
diff --git a/src/python/WMCore/WMBS/MySQL/Files/GetBulkRunLumi.py b/src/python/WMCore/WMBS/MySQL/Files/GetBulkRunLumi.py
index f00bce743..c9d240fc7 100644
--- a/src/python/WMCore/WMBS/MySQL/Files/GetBulkRunLumi.py
+++ b/src/python/WMCore/WMBS/MySQL/Files/GetBulkRunLumi.py
@@ -7,6 +7,7 @@ MySQL implementation of GetBulkRunLumi
 
 from WMCore.Database.DBFormatter import DBFormatter
 
+
 class GetBulkRunLumi(DBFormatter):
     """
     Note that this is ID based.  I may have to change it back
@@ -18,7 +19,7 @@ class GetBulkRunLumi(DBFormatter):
                WHERE flr.fileid = :id
     """
 
-    def getBinds(self, files = None):
+    def getBinds(self, files=None):
         binds = []
         files = self.dbi.makelist(files)
         for f in files:
@@ -32,19 +33,17 @@ class GetBulkRunLumi(DBFormatter):
         res = self.formatDict(result)
 
         for entry in res:
-            id  = entry['id']
+            fileid = entry['id']
             run = entry['run']
-            if not id in finalResult.keys():
-                finalResult[id] = {}
-            if not run in finalResult[id].keys():
-                finalResult[id][run] = []
-            finalResult[id][run].append(entry['lumi'])
+            finalResult.setdefault(fileid, {})
+            finalResult[fileid].setdefault(run, [])
+            finalResult[fileid][run].append(entry['lumi'])
 
         return finalResult
 
-    def execute(self, files = None, conn = None, transaction = False):
+    def execute(self, files=None, conn=None, transaction=False):
         binds = self.getBinds(files)
 
         result = self.dbi.processData(self.sql, binds,
-                         conn = conn, transaction = transaction)
+                                      conn=conn, transaction=transaction)
         return self.format(result)
diff --git a/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py b/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py
index 603923f05..68e3a7074 100644
--- a/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py
+++ b/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py
@@ -387,7 +387,9 @@ class DataCollectionService_t(unittest.TestCase):
                           u'lfn': u'/store/unmerged/fileB.root',
                           u'locations': [u'T1_US_FNAL_MSS'],
                           u'parents': ['file2'],
-                          u'runs': [{u'lumis': [4, 5, 6], u'run_number': 1}, {u'lumis': [9], u'run_number': 1}]},
+                          u'runs': [{u'lumis': [4, 5, 6], u'run_number': 1},
+                                    {u'lumis': [9], u'run_number': 1},
+                                    {u'lumis': [7, 8], u'run_number': 7}]},
                          {u'events': 10,
                           u'lfn': u'/store/unmerged/fileC.root',
                           u'locations': [u'T1_US_FNAL_Disk', u'T2_US_MIT'],
@@ -404,18 +406,26 @@ class DataCollectionService_t(unittest.TestCase):
             if item['lfn'] == '/store/unmerged/fileA.root':
                 self.assertEqual(item['events'], 165)
                 self.assertItemsEqual(item['locations'], ['T2_CH_CERN', 'T2_CH_CERNBOX'])
-                self.assertItemsEqual(item['runs'], [{'lumis': [1810823], 'run_number': 1},
-                                                     {'lumis': [1810823], 'run_number': 2},
-                                                     {'lumis': [1810824], 'run_number': 1}])
+                for runLumi in item['runs']:
+                    if runLumi['run_number'] == 1:
+                        self.assertItemsEqual(runLumi['lumis'], [1810824, 1810823])
+                    elif runLumi['run_number'] == 2:
+                        self.assertItemsEqual(runLumi['lumis'], [1810823])
+                    else:
+                        raise AssertionError("This should never happen")
                 self.assertEqual(item['parents'], [])
             elif item['lfn'] == '/store/unmerged/fileB.root':
                 self.assertEqual(item['events'], 50)
                 self.assertItemsEqual(item['locations'], ['T1_US_FNAL_MSS'])
-                self.assertItemsEqual(item['runs'], [{'lumis': [1, 2, 3], 'run_number': 1},
-                                                     {'lumis': [4, 5, 6], 'run_number': 1},
-                                                     {'lumis': [9], 'run_number': 1}])
+                for runLumi in item['runs']:
+                    if runLumi['run_number'] == 1:
+                        self.assertItemsEqual(runLumi['lumis'], [1, 2, 3, 4, 5, 6, 9])
+                    elif runLumi['run_number'] == 7:
+                        self.assertItemsEqual(runLumi['lumis'], [7, 8])
+                    else:
+                        raise AssertionError("This should never happen")
                 self.assertItemsEqual(item['parents'], ['file1', 'file2', 'file3'])
-            elif item['lfn'] == '/store/unmerged/fileB.root':
+            elif item['lfn'] == '/store/unmerged/fileC.root':
                 self.assertEqual(item['events'], 10)
                 self.assertItemsEqual(item['locations'], ['T1_US_FNAL_Disk', 'T2_US_MIT'])
                 self.assertItemsEqual(item['runs'], [{'lumis': [111], 'run_number': 222}])