feat: automatic job grouping by Stellatsuu · Pull Request #95 · DIRACGrid/dirac-cwl

Stellatsuu · 2026-02-03T16:19:11Z

Closes: #66
Related to: #61

Changes:

added input_data: list[str | File] to TransformationSubmissionModel and ProductionSubmissionModel
added optional inputs-file parameter to Transformation and Production CLIs:
dirac-cwl transformation/production submit file.cwl --inputs-file file.yaml
renamed parameter-path to input_files: list[str] in Job CLI:
dirac-cwl job submit file.cwl --input-files file1.yaml file2.yaml ...
added group_size executionHooksHint to Transformation Workflows, such as:

hints:
  - class: dirac:ExecutionHooks
    group_size: (int)

group_size determines the number of jobs to be created and how many inputs files they will contain in submit_transformation_router, by default, it equals 1, which mean a job will be created for each input in the inputs file. Once the list of jobs is created, it is sent to the job_router and processed.
added simple tests and workflows (e.g: count and list the inputs files contained in each created job)
quick fix about JobWrapper related tests: task.cwl was created during post_process but never cleared after running tests.

TODO after this PR:

plan an issue/PR for input sandbox within Transformations if not already done (e.g: feat: automatic job grouping #95 (comment))

src/dirac_cwl/transformation/__init__.py

test/test_workflows.py

test/workflows/automatic_job_grouping/inputs_files-strings.yaml

aldbr · 2026-02-06T11:44:36Z

src/dirac_cwl/submission_models.py

 class TransformationSubmissionModel(BaseModel):
    """Transformation definition sent to the router."""

    # Allow arbitrary types to be passed to the model
    model_config = ConfigDict(arbitrary_types_allowed=True)

    task: CommandLineTool | Workflow | ExpressionTool
+    input_data: Optional[list[str | File] | None] = None


As we are going to integrate input sandbox within transformations (#92), it would be interesting to see if we could reuse the JobInputModel (renamed as InputModel?)

Regarding @arrabito comments:

I agree that we don't need to have input sandbox for now, so it can't be local files.

I don't remember how we will add support for sandboxes in the transformation system. For simplicity, I would keep just LFN paths for now.

As said before, in my opinion there is no need to support/create sandboxes for now.

Do I still make this change in this PR? Or wouldn't it be better to do it in a (futur) sandbox PR? Maybe I missunderstood what you meant here.

Let's make this change in a future sandbox PR I would say

Thinking a little bit further, we may also want to allow local file paths, but only to be used for Local execution (without adding them to SB).

So if the submission is local we allow only local paths, while if the submission is to DIRAC we allow only LFN paths.

In this way, we could also execute transformations locally.

Eventually later on, we will also allow local file paths for DIRAC submission (adding them to ISB).

@aldbr what do you think?

src/dirac_cwl_proto/transformation/__init__.py

Stellatsuu · 2026-02-10T13:39:10Z

@aldbr Regarding this part of the code:

dirac-cwl/src/dirac_cwl_proto/transformation/__init__.py

Lines 130 to 163 in 72956d5

    
           # Temporary comment 
        
           # if transformation_execution_hooks.configuration and transformation_execution_hooks.group_size: 
        
           #     # Get the metadata class 
        
           #     transformation_metadata = transformation_execution_hooks.to_runtime(transformation) 
        
           # 
        
           #     # Build the input cwl for the jobs to submit 
        
           #     logger.info("Getting the input data for the transformation...") 
        
           #     input_data_dict = {} 
        
           #     min_length = None 
        
           #     for input_name, group_size in transformation_execution_hooks.group_size.items(): 
        
           #         # Get input query 
        
           #         logger.info("\t- Getting input query for %s...", input_name) 
        
           #         input_query = transformation_metadata.get_input_query(input_name) 
        
           #         if not input_query: 
        
           #             raise RuntimeError("Input query not found.") 
        
           # 
        
           #         # Wait for the input to be available 
        
           #         logger.info("\t- Waiting for input data for %s...", input_name) 
        
           #         logger.debug("\t\t- Query: %s", input_query) 
        
           #         logger.debug("\t\t- Group Size: %s", group_size) 
        
           #         while not (inputs := _get_inputs(input_query, group_size)): 
        
           #             logger.debug("\t\t- Result: %s", inputs) 
        
           #             time.sleep(5) 
        
           #         logger.info("\t- Input data for %s available.", input_name) 
        
           #         if not min_length or len(inputs) < min_length: 
        
           #             min_length = len(inputs) 
        
           # 
        
           #         # Update the input data in the metadata 
        
           #         # Only keep the first min_length inputs 
        
           #         input_data_dict[input_name] = inputs[:min_length] 
        
           # 
        
           #     # Get the JobModelParameter for each input 
        
           #     job_model_params = _generate_job_model_parameter(input_data_dict) 
        
           #     logger.info("Input data for the transformation retrieved!")

Are we planning on keeping it? Just so I un-comment it and make the changes related to the group_size type change.

Stellatsuu · 2026-02-10T13:41:30Z

Waiting on #66 (comment) and #95 (comment) approval about what we're doing, and then, PR should be ready to be fully reviewed (and potentially merged 🙏).

aldbr · 2026-02-10T13:45:45Z

@aldbr Regarding this part of the code:

dirac-cwl/src/dirac_cwl_proto/transformation/__init__.py

Lines 130 to 163 in 72956d5

# Temporary comment

# if transformation_execution_hooks.configuration and transformation_execution_hooks.group_size:

# # Get the metadata class

# transformation_metadata = transformation_execution_hooks.to_runtime(transformation)

#

# # Build the input cwl for the jobs to submit

# logger.info("Getting the input data for the transformation...")

# input_data_dict = {}

# min_length = None

# for input_name, group_size in transformation_execution_hooks.group_size.items():

# # Get input query

# logger.info("\t- Getting input query for %s...", input_name)

# input_query = transformation_metadata.get_input_query(input_name)

# if not input_query:

# raise RuntimeError("Input query not found.")

#

# # Wait for the input to be available

# logger.info("\t- Waiting for input data for %s...", input_name)

# logger.debug("\t\t- Query: %s", input_query)

# logger.debug("\t\t- Group Size: %s", group_size)

# while not (inputs := _get_inputs(input_query, group_size)):

# logger.debug("\t\t- Result: %s", inputs)

# time.sleep(5)

# logger.info("\t- Input data for %s available.", input_name)

# if not min_length or len(inputs) < min_length:

# min_length = len(inputs)

#

# # Update the input data in the metadata

# # Only keep the first min_length inputs

# input_data_dict[input_name] = inputs[:min_length]

#

# # Get the JobModelParameter for each input

# job_model_params = _generate_job_model_parameter(input_data_dict)

# logger.info("Input data for the transformation retrieved!")

Are we planning on keeping it? Just so I un-comment it and make the changes related to the group_size type change.

Yes we want to keep it. A transformation should either get inputs from the CLI, or from a DataCatalog/Bookkeeping service.

Stellatsuu · 2026-02-10T13:49:36Z

I’m also not sure whether the job_grouping workflow is worth keeping. It only counts the file in each job and list them, based on the inputs_file content.
I'm not sure on what to test otherwise than that.

Also, the test_run_transformation_with_inputs_file test only check if the transformation succeed, it doesn't check if the number of groups and jobs created are correct, maybe I should change that? If so, I think it would be better to create a dedicated test for this case, so we still have a more general test to see if transformation with inputs_file succeed or not and a dedicated test for this automatic grouping check.

If you have any ideas.

Stellatsuu · 2026-02-10T13:57:10Z

@aldbr Regarding this part of the code:

dirac-cwl/src/dirac_cwl_proto/transformation/__init__.py

Lines 130 to 163 in 72956d5

# Temporary comment

# if transformation_execution_hooks.configuration and transformation_execution_hooks.group_size:

# # Get the metadata class

# transformation_metadata = transformation_execution_hooks.to_runtime(transformation)

#

# # Build the input cwl for the jobs to submit

# logger.info("Getting the input data for the transformation...")

# input_data_dict = {}

# min_length = None

# for input_name, group_size in transformation_execution_hooks.group_size.items():

# # Get input query

# logger.info("\t- Getting input query for %s...", input_name)

# input_query = transformation_metadata.get_input_query(input_name)

# if not input_query:

# raise RuntimeError("Input query not found.")

#

# # Wait for the input to be available

# logger.info("\t- Waiting for input data for %s...", input_name)

# logger.debug("\t\t- Query: %s", input_query)

# logger.debug("\t\t- Group Size: %s", group_size)

# while not (inputs := _get_inputs(input_query, group_size)):

# logger.debug("\t\t- Result: %s", inputs)

# time.sleep(5)

# logger.info("\t- Input data for %s available.", input_name)

# if not min_length or len(inputs) < min_length:

# min_length = len(inputs)

#

# # Update the input data in the metadata

# # Only keep the first min_length inputs

# input_data_dict[input_name] = inputs[:min_length]

#

# # Get the JobModelParameter for each input

# job_model_params = _generate_job_model_parameter(input_data_dict)

# logger.info("Input data for the transformation retrieved!")

Are we planning on keeping it? Just so I un-comment it and make the changes related to the group_size type change.

Yes we want to keep it. A transformation should either get inputs from the CLI, or from a DataCatalog/Bookkeeping service.

Since group_size is now an int, this code is kinda broken now, no? I need an input_name to retrieve the input_query: input_query = transformation_metadata.get_input_query(input_name), where would this value be now? In a Transformation hint? As a list of input_names?

arrabito · 2026-02-10T15:43:59Z

Since group_size is now an int, this code is kinda broken now, no? I need an input_name to retrieve the input_query: input_query = transformation_metadata.get_input_query(input_name), where would this value be now? In a Transformation hint? As a list of input_names?

As far as I see, I'm not sure that any input_name is needed anymore.

In the current QueryBasedPlugin, input_name is just used to build the LFN path, see:

https://github.com/DIRACGrid/dirac-cwl/blob/main/src/dirac_cwl/execution_hooks/plugins/core.py#L37C31-L37C41

Probably we could just change get_input_query to not take any argument and just build LFN path as:

/query_root/vo/campaign/site/data_type

instead of:

/query_root/vo/campaign/site/data_type/input_name

Then, I guess that the group_size in yaml file should be specified as:

- class: dirac:ExecutionHooks
  hook_plugin: QueryBasedPlugin
  group_size: 5

instead of:

- class: dirac:ExecutionHooks
  hook_plugin: QueryBasedPlugin
  group_size:
    input-data: 5

@aldbr do you agree?

(Maybe some other changes are needed that I haven't thought).

test/test_integration.py

test/workflows/job_grouping/job_grouping.cwl

aldbr · 2026-02-10T17:02:52Z

Since group_size is now an int, this code is kinda broken now, no? I need an input_name to retrieve the input_query: input_query = transformation_metadata.get_input_query(input_name), where would this value be now? In a Transformation hint? As a list of input_names?

As far as I see, I'm not sure that any input_name is needed anymore.

In the current QueryBasedPlugin, input_name is just used to build the LFN path, see:

https://github.com/DIRACGrid/dirac-cwl/blob/main/src/dirac_cwl/execution_hooks/plugins/core.py#L37C31-L37C41

Probably we could just change get_input_query to not take any argument and just build LFN path as:

/query_root/vo/campaign/site/data_type

instead of:

/query_root/vo/campaign/site/data_type/input_name

Then, I guess that the group_size in yaml file should be specified as:
- class: dirac:ExecutionHooks
  hook_plugin: QueryBasedPlugin
  group_size: 5
instead of:
- class: dirac:ExecutionHooks
  hook_plugin: QueryBasedPlugin
  group_size:
    input-data: 5
@aldbr do you agree?

(Maybe some other changes are needed that I haven't thought).

Yes I agree. In any case, this is going to be revised at some point with the hints proposed in #69

Stellatsuu · 2026-02-18T10:16:27Z

Current PR status:

waiting on more info about Support data-processing transformations specified through a static list of input files #66 (comment) (@arrabito @aldbr)
need to find a way to use fixtures for JobWrapper related tests (currently dealing with "coroutine errors"): feat: automatic job grouping #95 (comment) (must be because the tests aren't using our mocks)
need a review @aldbr

Stellatsuu · 2026-03-03T09:26:42Z

I don't know how to fix current lint error. I had to rebase and reword all my "wrong" commits and it's still not working because of old commits already pushed.

Stellatsuu · 2026-03-03T09:30:59Z

Also, PyPi is failling on something I didn't touch directly (mostly sure about that), so I don't really know what to do about that too. If you have any ideas

scripts/generate_schemas.py:70: error: Dict entry 0 has incompatible type "str": "type[JobInputModel]"; expected "str": "type[BaseModel]"  [dict-item]
Found 1 error in 1 file (checked 54 source files)

 models.update(
            {
                "JobInputModel": JobInputModel, # <--- error happens here
                "JobSubmissionModel": JobSubmissionModel,
                "TransformationSubmissionModel": TransformationSubmissionModel,
                "ProductionSubmissionModel": ProductionSubmissionModel,
            }
        )

class JobInputModel(BaseModel): # <-- it's a BaseModel ?
    """Input data and sandbox files for a job execution."""

    # Allow arbitrary types to be passed to the model
    model_config = ConfigDict(arbitrary_types_allowed=True)

    sandbox: list[str] | None
    cwl: dict[str, Any]

    @field_serializer("cwl")
    def serialize_cwl(self, value):
        """Serialize CWL object to dictionary.

        :param value: CWL object to serialize.
        :return: Serialized CWL dictionary.
        """
        return save(value)

src/dirac_cwl/transformation/__init__.py

src/dirac_cwl/production/__init__.py

src/dirac_cwl/transformation/__init__.py

test/test_workflows.py

natthan-pigoux · 2026-03-04T14:11:48Z

Thank you @Stellatsuu , apart the small comments the overall looks good to me.

natthan-pigoux · 2026-03-06T13:15:23Z

src/dirac_cwl/submission_models.py

            raise TypeError(f"Cannot serialize type {type(value)}")

+    @field_serializer("input_data")
+    def serialize_input_data(self, value):


I see that the TransformationSubmissionModel and the ProductionSubmissionModel share methods and arguments from now. I don't know if we want to create a BaseModel for them to inherit from. Like:

class ???SubmissionModel(BaseModel): """BaseModel for high level DIRAC Workflow""" # Allow arbitrary types to be passed to the model model_config = ConfigDict(arbitrary_types_allowed=True) task: CommandLineTool | Workflow | ExpressionTool input_data: Optional[list[str | File] | None] = None @field_serializer("task") def serialize_task(self, value): """Serialize CWL task object to dictionary. :param value: CWL task object to serialize. :return: Serialized task dictionary. :raises TypeError: If value is not a valid CWL task type. """ if isinstance(value, (CommandLineTool, Workflow, ExpressionTool)): return save(value) else: raise TypeError(f"Cannot serialize type {type(value)}") @field_serializer("input_data") def serialize_input_data(self, value): """Serialize an input data list to a list of strings. :param value: Input data list to serialize. :return: Serialized input data list. """ if value: return [save(item) if isinstance(item, File) else item for item in value] return None

@aldbr do you think it is worth it? Or this will evolve soon and this will not be the case afterward.

Stellatsuu temporarily deployed to github-pages February 3, 2026 16:19 — with GitHub Actions Inactive

aldbr requested changes Feb 6, 2026

View reviewed changes

Stellatsuu temporarily deployed to github-pages February 10, 2026 13:30 — with GitHub Actions Inactive

Stellatsuu temporarily deployed to github-pages February 10, 2026 13:36 — with GitHub Actions Inactive

aldbr requested changes Feb 10, 2026

View reviewed changes

test/test_integration.py Outdated Show resolved Hide resolved

test/workflows/job_grouping/job_grouping.cwl Show resolved Hide resolved

Stellatsuu temporarily deployed to github-pages February 11, 2026 11:49 — with GitHub Actions Inactive

Stellatsuu self-assigned this Feb 17, 2026

Stellatsuu temporarily deployed to github-pages February 19, 2026 12:20 — with GitHub Actions Inactive

Stellatsuu temporarily deployed to github-pages February 19, 2026 12:56 — with GitHub Actions Inactive

aldbr requested a review from natthan-pigoux March 2, 2026 16:48

Stellatsuu temporarily deployed to github-pages March 3, 2026 08:37 — with GitHub Actions Inactive

Stellatsuu temporarily deployed to github-pages March 3, 2026 09:23 — with GitHub Actions Inactive

Stellatsuu temporarily deployed to github-pages March 3, 2026 15:24 — with GitHub Actions Inactive

Stellatsuu closed this Mar 3, 2026

Stellatsuu force-pushed the automatic-job-grouping branch from f9337a8 to 52e6144 Compare March 3, 2026 15:26

Stellatsuu reopened this Mar 3, 2026

Stellatsuu temporarily deployed to github-pages March 3, 2026 15:28 — with GitHub Actions Inactive

Stellatsuu temporarily deployed to github-pages March 3, 2026 15:29 — with GitHub Actions Inactive

Stellatsuu marked this pull request as ready for review March 3, 2026 15:29

natthan-pigoux reviewed Mar 4, 2026

View reviewed changes

src/dirac_cwl/transformation/__init__.py Show resolved Hide resolved

natthan-pigoux requested changes Mar 4, 2026

View reviewed changes

src/dirac_cwl/production/__init__.py Outdated Show resolved Hide resolved

src/dirac_cwl/production/__init__.py Outdated Show resolved Hide resolved

natthan-pigoux requested changes Mar 4, 2026

View reviewed changes

src/dirac_cwl/transformation/__init__.py Show resolved Hide resolved

natthan-pigoux requested changes Mar 4, 2026

View reviewed changes

test/test_workflows.py Outdated Show resolved Hide resolved

Stellatsuu temporarily deployed to github-pages March 4, 2026 23:34 — with GitHub Actions Inactive

Stellatsuu temporarily deployed to github-pages March 6, 2026 11:50 — with GitHub Actions Inactive

Stellatsuu added 17 commits March 6, 2026 12:53

feat: add input_data to TransformationSubmissionModel, updated CLI

b12d6fa

chore: add test workflows for transformation grouping

5a92395

feat: added Job Grouping

00cdfcd

chore: rename

1dc091c

chore: added job grouping test

c73532d

fix: updated transformation hoooks group_size

5dd32de

refactor: rename

a939b0a

refactor: updated tests

987c099

refactor: updated group_size definition and values

047ceaa

fix: fixed JobWrapper tests where task.cwl was not deleted

74c1dfa

refactor: removed input_name from get_input_query

7a5bd3f

fix: file location

672cfdc

feat: add input_data to ProductionSubmissionModel

1053ba3

fix: fixed tests and jobwrapper imports

e15dade

fix: fixed imports

fe50e18

fix: removed pytest warnings

a3582cc

fix: fixed test

3a1314c

Stellatsuu force-pushed the automatic-job-grouping branch from 99df614 to 3a1314c Compare March 6, 2026 11:54

Stellatsuu temporarily deployed to github-pages March 6, 2026 11:55 — with GitHub Actions Inactive

fix: fixed mypy error (?)

9a4bae3

Stellatsuu temporarily deployed to github-pages March 6, 2026 12:11 — with GitHub Actions Inactive

Stellatsuu requested review from aldbr and natthan-pigoux March 6, 2026 12:15

natthan-pigoux reviewed Mar 6, 2026

View reviewed changes

Conversation

Stellatsuu commented Feb 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

aldbr Feb 6, 2026

Choose a reason for hiding this comment

Uh oh!

Stellatsuu Feb 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aldbr Feb 10, 2026

Choose a reason for hiding this comment

Uh oh!

arrabito Feb 11, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Stellatsuu commented Feb 10, 2026

Uh oh!

Stellatsuu commented Feb 10, 2026

Uh oh!

aldbr commented Feb 10, 2026

Uh oh!

Stellatsuu commented Feb 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Stellatsuu commented Feb 10, 2026

Uh oh!

arrabito commented Feb 10, 2026

Uh oh!

Uh oh!

Uh oh!

aldbr commented Feb 10, 2026

Uh oh!

Stellatsuu commented Feb 18, 2026

Uh oh!

Stellatsuu commented Mar 3, 2026

Uh oh!

Stellatsuu commented Mar 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

natthan-pigoux commented Mar 4, 2026

Uh oh!

natthan-pigoux Mar 6, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Stellatsuu commented Feb 3, 2026 •

edited

Loading

Stellatsuu Feb 10, 2026 •

edited

Loading

Stellatsuu commented Feb 10, 2026 •

edited

Loading

Stellatsuu commented Mar 3, 2026 •

edited

Loading