Skip to content

[DCP Ingestion] Add preprocessing stage to workflow #177

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

Open
gmechali wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[DCP Ingestion] Add preprocessing stage to workflow #177

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 10 additions & 1 deletion infra/dcp/modules/ingestion/workflow/workflow.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ main:
- execution_error: null
- postprocessing_result: null
- embedding_result: null
- current_stage: "dataflow"
- current_stage: "preprocessing"

@gemini-code-assist gemini-code-assist Bot Jul 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Initializing current_stage to "preprocessing" is logical, but please note that the except block that uses current_stage to report failures is defined inside the process_ingestion step (lines 173-294). Since process_ingestion is only executed after lock acquisition, any failure during the actual preprocessing stage (which runs before lock acquisition) will not trigger that except block. Therefore, initializing current_stage here does not automatically protect the preprocessing stage from leaving the history in a RUNNING state upon failure.

- job_id: null
- lock_timeout: ${lock_acquisition_timeout}
- lock_retries: 0
Expand Down Expand Up @@ -96,6 +96,15 @@ main:
- sanitized_short_import: '$${text.substring(combined_import_name, 0, substring_end)}'
- dataflow_job_name: '$${"${clean_namespace_prefix}" + sanitized_short_import + "-" + string(int(sys.now()))}'

- update_history_preprocessing:
call: update_history
args:
workflow_id: '$${workflow_id}'
status: 'RUNNING'
stage: 'preprocessing'
job_id: null
import_list: '$${imports_history_list}'
Comment on lines +99 to +106

@gemini-code-assist gemini-code-assist Bot Jul 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Issue: Missing Error Handling for Preprocessing Stage

The new update_history_preprocessing step sets the ingestion history status to RUNNING for the preprocessing stage. However, the subsequent run_preprocessing step (and other setup steps like read_handshake_file) are executed outside of any try-catch block.

If the preprocessing Cloud Run job fails or if reading the handshake file fails, the workflow will terminate immediately. Because the global except block is only defined inside the process_ingestion step (which is executed much later, after lock acquisition), the workflow will never call update_history to mark the status as FAILURE. This will leave the ingestion permanently stuck in the RUNNING state in the history database and UI.

Recommendation

Wrap the preprocessing and handshake steps in a try-catch block to ensure that any failures during this stage are caught and the history is updated to FAILURE.

For example:

    - preprocessing_stage:
        try:
          steps:
            - update_history_preprocessing:
                call: update_history
                args:
                  workflow_id: '$${workflow_id}'
                  status: 'RUNNING'
                  stage: 'preprocessing'
                  job_id: null
                  import_list: '$${imports_history_list}'

            - run_preprocessing:
                call: launch_and_poll_preprocessing_job
                args:
                  job_name: '$${prep_job_name}'
                  workflow_id: '$${workflow_id}'
                  temp_location: '$${input.tempLocation}'
                  imports_arg: '$${comma_separated_imports}'
                  region: '$${input.region}'
                  project_id: '${project_id}'
                result: prep_res

            - read_handshake_file:
                call: googleapis.storage.v1.objects.get
                args:
                  bucket: '$${bucket_name}'
                  object: '$${text.url_encode(temp_path_prefix + "/datacommons/ingestion_records/" + workflow_id + ".json")}'
                  alt: 'media'
                result: handshake_data

            - update_launch_params:
                assign:
                  - launch_params.importList: '$${handshake_data.importList}'
                  - decoded_imports: '$${json.decode(handshake_data.importList)}'
        except:
          as: e
          steps:
            - update_history_preprocessing_failure:
                call: update_history
                args:
                  workflow_id: '$${workflow_id}'
                  status: 'FAILURE'
                  stage: 'preprocessing'
                  job_id: null
                  import_list: '$${imports_history_list}'
            - raise_preprocessing_error:
                raise: $${e}


- run_preprocessing:
call: launch_and_poll_preprocessing_job
args:
Expand Down
Loading