[DCP Ingestion] Update the ingestion workflow to use the new ingestion history updates.

[gmechali]

No description provided.

[gemini-code-assist]

Code Review

This pull request updates the ingestion orchestrator Google Workflow to track and update ingestion history at various stages (PENDING, RUNNING, SUCCESS, and FAILURE) using a new update_history sub-workflow. The feedback focuses on ensuring runtime safety by initializing job_id to prevent errors in the exception handler, adhering to the single import limitation by using import_name instead of import_list, improving step naming clarity, and wrapping workflow expressions in single quotes for consistency and parsing safety.

[gemini-code-assist]

The variable job_id is not initialized in the init step. If an exception occurs before the get_job_id step (for example, during set_import_staging or run_flex_template), the except block will attempt to evaluate default(job_id, null). Since job_id is not defined in the runtime scope at that point, this will raise a runtime error, masking the original failure and preventing the ingestion history from being updated.

Initializing job_id to null in the init step resolves this issue.

            - version: '$${"version-" + string(int(sys.now()))}'
            - bucket_name: '$${text.split(input.tempLocation, "/")[2]}'
            - execution_error: null
            - current_stage: "dataflow"
            - job_id: null
            - lock_timeout: ${var.lock_acquisition_timeout}
            - run_embeddings: ${var.enable_embeddings_generation}
            - run_postproc: ${var.enable_bigquery_postprocessing}

[gemini-code-assist]

Since job_id is now initialized to null in the init step, it is guaranteed to be defined in the runtime scope. Therefore, you can simplify the job_id argument by passing it directly without the default() helper. Additionally, ensure we use import_name with input.importName instead of import_list to adhere to the single import limitation.

              - update_history_failure:
                  call: update_history
                  args:
                    workflow_id: '$${workflow_id}'
                    status: 'FAILURE'
                    stage: '$${current_stage}'
                    job_id: '$${job_id}'
                    import_name: '$${input.importName}'

References

1. The ingestion workflow currently supports only a single import. Ensure that any changes are consistent with this limitation, using input.importName rather than input.importList.

[gemini-code-assist]

The step update_history_postproc_success is executed at the end of the successful workflow regardless of whether postprocessing was actually run (it is skipped if local.should_run_postprocessing is false). Naming it update_history_postproc_success is misleading. Renaming it to update_history_success or update_history_completed would be more accurate and maintainable. Also, ensure we use import_name instead of import_list to align with the single import limitation.

              - update_history_success:
                  call: update_history
                  args:
                    workflow_id: '$${workflow_id}'
                    status: 'SUCCESS'
                    stage: null
                    job_id: null
                    import_name: null

References

1. The ingestion workflow currently supports only a single import. Ensure that any changes are consistent with this limitation, using input.importName rather than input.importList.

[gemini-code-assist]

For consistency with the rest of the workflow definition and to prevent potential YAML parsing issues in Google Cloud Workflows (especially if any of these variables contain colons or other special characters), wrap the expressions in single quotes. Additionally, use importName with input.importName instead of importList to remain consistent with the single import limitation.

            body:
              workflowId: '$${workflow_id}'
              status: '$${status}'
              stage: '$${stage}'
              jobId: '$${job_id}'
              importName: '$${input.importName}'

References

1. The ingestion workflow currently supports only a single import. Ensure that any changes are consistent with this limitation, using input.importName rather than input.importList.