Use BigQuery instead of Spanner for import state

[vish-cs]

Summary

Migrates import-automation/helper import state management (ImportHistory and ImportSummary) from Cloud Spanner to BigQuery using an event-sourced append-only table + view architecture.

Key Changes

• BigQuery Schema (clients/schema.sql):
  • Replaced Spanner DDL with an append-only ImportHistory table (partitioned by DATE(UpdateTimestamp), clustered by ImportName) and a logical ImportSummary SQL view (QUALIFY ROW_NUMBER() OVER (PARTITION BY ImportName ORDER BY UpdateTimestamp DESC) = 1).
  • Removed redundant/unused columns (GraphPath and duplicate WorkflowExecutionID/WorkflowId in favor of JobId) while preserving DataImportTimestamp and NextRefreshTimestamp.
• BigQuery Client (clients/bigquery.py):
  • Implemented BigQueryClient using streaming inserts (insert_rows_json) to avoid BigQuery DML lock contention on concurrent import status updates, with automatic dataset/schema initialization on NotFound.
• Service Routes & Config (routes/, config.py, dependencies.py, cloudbuild.yaml):
  • Updated /imports/status, /imports/version, /imports/feed, and /database/initialize to use BigQueryClient.
  • Configured BQ_DATASET_ID=import_automation for prod (import-helper-service) and BQ_DATASET_ID=import_automation_staging for staging/CI (import-helper-service-staging).
• Tests (app_test.py):
  • Updated endpoint unit tests to use get_bigquery_client and added unit tests for BigQueryClient initialization and streaming writes.

[gemini-code-assist]

Code Review

This pull request migrates the Data Commons Import Helper Service database backend from Cloud Spanner to BigQuery, introducing a new BigQueryClient and updating associated routes, tests, configuration, and deployment files. The database schema is defined via an append-only ImportHistory partitioned table and an ImportSummary view. Feedback on these changes suggests avoiding global logger configuration side effects in the client module, executing the SQL schema initialization script in a single query call rather than manually splitting by semicolons, and simplifying the DataImportTimestamp window function expression in the SQL view definition.

[gemini-code-assist]

Setting the root logger's level (logging.getLogger().setLevel(logging.INFO)) inside a library or client module is a side effect that affects the entire application. It can override the logging configuration set by the web server (like Uvicorn) or the main application entry point. It is best to configure logging levels at the application startup (e.g., in app.py) and simply use logging.getLogger(__name__) or standard logging.info within module files.

[gemini-code-assist]

BigQuery supports executing multiple SQL statements (scripts) in a single query call. Splitting the schema by ; manually is fragile and can easily break if a semicolon is present inside a comment or a string literal in the SQL file. You can execute the entire rendered_schema in a single query call, which is safer and more efficient.

        if rendered_schema.strip():
            self.client.query(rendered_schema).result()

[gemini-code-assist]

Since DataImportTimestamp is only populated (with UpdateTimestamp) when Status = 'STAGING' and is NULL otherwise, the expression MAX(IF(Status = 'STAGING', UpdateTimestamp, DataImportTimestamp)) can be simplified to just MAX(DataImportTimestamp). This improves readability and reduces complexity in the view definition.

Suggested change

	MAX(IF(Status = 'STAGING', UpdateTimestamp, DataImportTimestamp)) OVER (
	PARTITION BY ImportName
	) AS DataImportTimestamp,
	MAX(DataImportTimestamp) OVER (
	PARTITION BY ImportName
	) AS DataImportTimestamp,