Add rollback to ingestion workflow

[vish-cs]

• Update spanner ingestion workflow to add a rollback sub-workflow to handle failures
• Remove runIngestion parameter to always invoke spanner ingestion workflow from the import automation workflow (batch job)
• Refactoring/clean up of the workflow

[codacy-production]

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

View in Codacy

🟢 Metrics 0 complexity

Metric	Results
Complexity	0

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.}

[gemini-code-assist]

Code Review

This pull request refactors the import automation and Spanner ingestion workflows to dynamically resolve import payloads, simplify ingestion triggering, and introduce structured sub-workflows for state updates, pipeline execution, and rollback on failure. Additionally, SQL query optimization in the provenance summary generator reduces the IN filter threshold to 20KB with a fallback to stream all node names. Feedback is provided regarding a potential runtime crash in the Spanner ingestion workflow when attempting to retrieve the job ID from a non-map error object during a pipeline failure, along with a suggested try/except block to safely resolve it.

[gemini-code-assist]

Using map.get(pipeline_err, "job_id") can cause a runtime crash if pipeline_err is a string (e.g., if an HTTP request fails or a system error occurs) rather than a map. In Google Cloud Workflows, calling map.get on a non-map type throws a TypeError, which would crash the except block itself, preventing the lock from being released and leaving the workflow in an inconsistent state.

To make this robust, we can safely determine the job_id using a try/except step block before calling rollback_subworkflow.

            - determine_job_id:
                try:
                  assign:
                    - failed_job_id: ${pipeline_err.job_id}
                except:
                  as: err
                  assign:
                    - failed_job_id: "N/A"
            - run_rollback:
                call: rollback_subworkflow
                args:
                  workflow_id: '${workflow_id}'
                  helper_url: ${helper_url}
                  import_list: ${import_info_list}
                  project_id: ${project_id}
                  dataflow_job_name: ${dataflow_job_name}
                  dataflow_gcs_path: ${dataflow_gcs_path}
                  location: ${location}
                  spanner_project_id: ${spanner_project_id}
                  spanner_instance_id: ${spanner_instance_id}
                  spanner_database_id: ${spanner_database_id}
                  wait_period: ${wait_period}
                  job_id: ${failed_job_id}
                  stages: ${stages}