Conversation
There was a problem hiding this comment.
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.
| from google.api_core import exceptions as gcp_exceptions | ||
| from google.cloud import bigquery | ||
|
|
||
| logging.getLogger().setLevel(logging.INFO) |
There was a problem hiding this comment.
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.
| for raw_stmt in rendered_schema.split(";"): | ||
| lines = [ | ||
| line for line in raw_stmt.splitlines() | ||
| if not line.strip().startswith("--") | ||
| ] | ||
| cleaned = "\n".join(lines).strip() | ||
| if cleaned: | ||
| self.client.query(cleaned).result() |
There was a problem hiding this comment.
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()| MAX(IF(Status = 'STAGING', UpdateTimestamp, DataImportTimestamp)) OVER ( | ||
| PARTITION BY ImportName | ||
| ) AS DataImportTimestamp, |
There was a problem hiding this comment.
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.
| MAX(IF(Status = 'STAGING', UpdateTimestamp, DataImportTimestamp)) OVER ( | |
| PARTITION BY ImportName | |
| ) AS DataImportTimestamp, | |
| MAX(DataImportTimestamp) OVER ( | |
| PARTITION BY ImportName | |
| ) AS DataImportTimestamp, |
0458c0f to
3330eac
Compare
Summary
Migrates
import-automation/helperimport state management (ImportHistoryandImportSummary) from Cloud Spanner to BigQuery using an event-sourced append-only table + view architecture.Key Changes
clients/schema.sql):ImportHistorytable (partitioned byDATE(UpdateTimestamp), clustered byImportName) and a logicalImportSummarySQL view (QUALIFY ROW_NUMBER() OVER (PARTITION BY ImportName ORDER BY UpdateTimestamp DESC) = 1).GraphPathand duplicateWorkflowExecutionID/WorkflowIdin favor ofJobId) while preservingDataImportTimestampandNextRefreshTimestamp.clients/bigquery.py):BigQueryClientusing streaming inserts (insert_rows_json) to avoid BigQuery DML lock contention on concurrent import status updates, with automatic dataset/schema initialization onNotFound.routes/,config.py,dependencies.py,cloudbuild.yaml):/imports/status,/imports/version,/imports/feed, and/database/initializeto useBigQueryClient.BQ_DATASET_ID=import_automationfor prod (import-helper-service) andBQ_DATASET_ID=import_automation_stagingfor staging/CI (import-helper-service-staging).app_test.py):get_bigquery_clientand added unit tests forBigQueryClientinitialization and streaming writes.