DM-55338: Implement staging of update records#53
Open
JeremyMcCormick wants to merge 1 commit into
Open
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #53 +/- ##
==========================================
+ Coverage 81.13% 81.42% +0.29%
==========================================
Files 45 44 -1
Lines 2539 2504 -35
Branches 237 233 -4
==========================================
- Hits 2060 2039 -21
+ Misses 390 379 -11
+ Partials 89 86 -3 ☔ View full report in Codecov by Harness. |
This changes update records to be handled more like regular table data in the replication and data ingestion. Instead of the promotion process working directly from the parquet files in cloud storage, it now reads from a staging table that is built similarly to the other tables such as DiaObject and DiaSource.
JeremyMcCormick
force-pushed
the
tickets/DM-55338
branch
from
34b53f4 to
693dab0
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Implements DM-55338 by shifting update handling to a staged-raw-records model: raw APDB update events are staged into a BigQuery staging.updates table (from updates.parquet), then expanded into field-level rows in promotion.expanded_updates, reduced to a latest_only table, and merged into DIA target tables.
Changes:
- Replace
update_records.parquetwithupdates.parquetand persistapdb_replica_chunkper row inUpdateRecords. - Remove
UpdateRecordExpanderand centralize expansion logic inExpandedUpdateRecord.from_update_record, plusExpandedUpdatesTablefor expanded/latest-only tables. - Update
UpdatesManagerand tests to read raw staged updates from BigQuery, build expanded/latest-only tables, and merge into targets.
Reviewed changes
Copilot reviewed 19 out of 19 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| tests/test_updates_merger.py | Switch merger tests to use ExpandedUpdatesTable + ExpandedUpdateRecord to build latest-only inputs. |
| tests/test_updates_manager.py | Simulate Dataflow by loading updates.parquet into staging.updates before running UpdatesManager. |
| tests/test_update_records.py | Update tests for UpdateRecords.records now being (apdb_replica_chunk, record) pairs. |
| tests/test_expanded_updates_table.py | Rename/update tests for ExpandedUpdatesTable and move expanded updates to promotion dataset. |
| tests/test_expanded_update_record.py | Update tests to use ExpandedUpdateRecord.from_update_record expansion API. |
| tests/test_dataset_builder.py | Adjust expected tables to include raw updates table in staging dataset. |
| tests/test_cli.py | Update CLI dataset creation expectations (staging now has 4 tables). |
| tests/test_chunk_uploader.py | Use UpdateRecords.PARQUET_FILE_NAME in assertions/glob for updates parquet. |
| tests/test_chunk_promoter.py | Add staging of raw updates table; expand updates via ExpandedUpdateRecord.from_update_record. |
| python/lsst/dax/ppdb/tests/_updates.py | Test helper now returns UpdateRecords with per-record chunk id. |
| python/lsst/dax/ppdb/bigquery/updates/updates_manager.py | Rework manager to read staged raw updates from BigQuery and build expanded/latest-only tables in promotion. |
| python/lsst/dax/ppdb/bigquery/updates/update_records.py | Add apdb_replica_chunk column; rename parquet to updates.parquet; store (chunk, record) pairs. |
| python/lsst/dax/ppdb/bigquery/updates/update_record_expander.py | Remove legacy expander (superseded by ExpandedUpdateRecord.from_update_record). |
| python/lsst/dax/ppdb/bigquery/updates/expanded_updates_table.py | Rename/refactor table manager for expanded + latest-only tables; update schema column names. |
| python/lsst/dax/ppdb/bigquery/updates/expanded_update_record.py | Add apdb_replica_chunk field and from_update_record constructor. |
| python/lsst/dax/ppdb/bigquery/updates/init.py | Export new expanded table/record modules; stop exporting removed modules. |
| python/lsst/dax/ppdb/bigquery/schema/dataset_builder.py | Add staging.updates raw updates table schema for Dataflow staging. |
| python/lsst/dax/ppdb/bigquery/ppdb_bigquery.py | Write updates parquet as (replica_chunk.id, record) pairs. |
| python/lsst/dax/ppdb/bigquery/chunk_uploader.py | Adjust upload flow and Pub/Sub trigger logic around staging updates. |
Comments suppressed due to low confidence (1)
tests/test_expanded_update_record.py:283
test_empty_recordscurrently iterates over a hard-coded empty list (for ... in []), so it will pass regardless of how update records expansion behaves. It would be more robust to construct an emptyUpdateRecordsinstance and iterate overempty_update_records.recordsso the test actually validates the empty-input behavior.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+265
to
270
| # 3) Update the database record for the chunk, setting status | ||
| # to 'staged' and populating the GCS URI. | ||
| new_status = ChunkStatus.STAGED | ||
| updated_replica_chunk = replica_chunk.with_new_status(new_status).with_new_gcs_uri( | ||
| f"gs://{gcs_uri}" | ||
| ) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This changes the handling of the update records to be more similar to how the other table data is handled.