Use PITR based rollback pipeline

[vish-cs]

Update spanner-ingestion-workflow rollback logic to use PITR rollback pipeline

Summary

Updates spanner-ingestion-workflow.yaml and deploy-services.yaml to use the new Spanner Point-In-Time Recovery (PITR) Dataflow rollback pipeline (#748) instead of reverting import versions and re-running ingestion/aggregation on failure.

Changes

• Pre-Ingestion Snapshot Timestamp ($T_{\text{pre}}$): Captures rollback_timestamp (time.format(sys.now())) immediately after acquiring the global Spanner ingestion lock and before ingestion starts.
• Rollback Pipeline Execution: Updates rollback_subworkflow and run_dataflow_job to launch the RollbackPipeline Dataflow Flex Template (rollback-job-<timestamp>) with importList and rollbackTimestamp.
• State Tracking: Updates import status and workflow execution history to RETRY via update_state_subworkflow once the PITR rollback job completes, and logs both the original pipeline failure and rollback failure if rollback fails.
• Deployment Config: Passes ROLLBACK_TEMPLATE_PATH to spanner-ingestion-workflow in deploy-services.yaml with a safe fallback derived from DATAFLOW_TEMPLATE_PATH.

TAG=agy
CONV=d4c3965b-966c-44f5-a041-24325139356a

[codacy-production]

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

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 integrates Spanner Point-in-Time Recovery (PITR) rollback capabilities into the ingestion workflow by introducing a rollback Dataflow pipeline. Key changes include capturing a pre-ingestion timestamp and updating the rollback subworkflow to execute the rollback Dataflow job. The review feedback highlights critical improvements: first, defaulting dataflow_gcs_path to an empty string to prevent a runtime TypeError caused by eager evaluation of text.replace_all when the environment variable is missing; second, wrapping subworkflow parameters that have default values (such as rollback_timestamp and pipeline_err) in curly braces to adhere to Google Cloud Workflows syntax.

[gmechali]

I want to make sure that we don't release the lock if rollback fails. If it fails then we should continue to keep the ingestion lock and fire some kind of alert. Are you aligned with that behavior?

Other than that lgtm!