Skip to content

Add skip_empty_input parameter to insertFilesForDataset_JEDI #774

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tmaeno merged 2 commits into from
Aug 24, 2026
Merged

Add skip_empty_input parameter to insertFilesForDataset_JEDI #774

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
980d37e
Add skip_empty_input parameter to insertFilesForDataset_JEDI
tmaeno Aug 24, 2026
0444687
Merge branch 'master' of https://github.com/tmaeno/panda-server
tmaeno Aug 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions pandajedi/jedicore/JediTaskBuffer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ def insertFilesForDataset_JEDI(
order_by,
maxFileRecords,
skip_short_output,
skip_empty_input,
lfn_constituent_map=None,
):
with self.proxyPool.get() as proxy:
Expand Down Expand Up @@ -112,6 +113,7 @@ def insertFilesForDataset_JEDI(
order_by,
maxFileRecords,
skip_short_output,
skip_empty_input,
lfn_constituent_map=lfn_constituent_map,
)

Expand Down
6 changes: 6 additions & 0 deletions pandajedi/jediorder/ContentsFeeder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -533,6 +533,11 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
skip_short_output = True
else:
skip_short_output = False
# skip empty input
if datasetSpec.isMaster() and not datasetSpec.isPseudo() and taskParamMap.get("skipEmptyInput"):
skip_empty_input = True
else:
skip_empty_input = False
# order by
if taskSpec.order_input_by() and datasetSpec.isMaster() and not datasetSpec.isPseudo():
orderBy = taskSpec.order_input_by()
Expand Down Expand Up @@ -585,6 +590,7 @@ def feed_contents_to_tasks(self, task_ds_list, real_run=True):
orderBy,
maxFileRecords,
skip_short_output,
skip_empty_input,
lfn_constituent_map=lfn_constituent_map,
)
retDB = res_dict["ret_val"]
Expand Down
9 changes: 8 additions & 1 deletion pandaserver/taskbuffer/db_proxy_mods/task_complex_module.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,7 @@ def insertFilesForDataset_JEDI(
order_by,
maxFileRecords,
skip_short_output,
skip_empty_input,
lfn_constituent_map=None,
):
comment = " /* JediDBProxy.insertFilesForDataset_JEDI */"
Expand All @@ -222,7 +223,10 @@ def insertFilesForDataset_JEDI(
tmpLog.debug(f"len(fileMap)={len(fileMap)} pid={pid}")
tmpLog.debug(f"datasetState={datasetState} dataset.state={datasetSpec.state}")
tmpLog.debug(f"respectLB={respectLB} tgtNumEventsPerJob={tgtNumEventsPerJob} skipFilesUsedBy={skipFilesUsedBy} ramCount={ramCount}")
tmpLog.debug(f"skipShortInput={skipShortInput} skipShortOutput={skip_short_output} inputPreStaging={inputPreStaging} order_by={order_by}")
tmpLog.debug(
f"skipShortInput={skipShortInput} skipShortOutput={skip_short_output} skip_empty_input={skip_empty_input} "
f"inputPreStaging={inputPreStaging} order_by={order_by}"
)
# return value for failure
diagMap = {"errMsg": "", "nChunksForScout": nChunksForScout, "nActivatedPending": 0, "isRunningTask": False}
# failedRet = False, 0, None, diagMap
Expand Down Expand Up @@ -453,6 +457,9 @@ def insertFilesForDataset_JEDI(
tmpNumEvents = int(fileVal["events"])
except Exception:
pass
if skip_empty_input and tmpNumEvents == 0:
tmpLog.debug(f"skip {fileSpec.lfn} due to nEvents=0")
continue
if skipShortInput and tmpNumEvents is not None:
# set multiples of nEventsPerJob if actual nevents is small
if tmpNumEvents >= nEventsPerFile:
Expand Down