feat(aggregation): support cross-provenance linked edges with delta sync - #736
feat(aggregation): support cross-provenance linked edges with delta sync#736clincoln8 wants to merge 1 commit into
Conversation
- Implements global transitive closure generation for linkedContainedInPlace, linkedMemberOf, and linkedMember across all DB provenances. - Partitions generated edges into canonical domain singletons (generated/LinkedPlaces, generated/LinkedTopics, generated/LinkedSVGs). - Implements BigQuery symmetric diff (EXCEPT DISTINCT) and Spanner blind batch deletions with 4-part composite primary key (subject_id, predicate, object_id, provenance) in 5,000-key chunks. - Adds indexed Early-Return Guard using EdgeByProvenance with LIMIT 1. - Introduces --enable_global_linked_edges flag (default: False) to dynamically route LINKED_EDGES to Phase 2 global execution in DCP while remaining a 100% no-op in legacy/Base DC execution. - Added comprehensive unit tests (84/84 passing).
Not up to standards ⛔🔴 Issues
|
| Category | Results |
|---|---|
| UnusedCode | 1 minor |
| ErrorProne | 1 high |
| Security | 5 medium |
| CodeStyle | 1 minor |
| Complexity | 1 medium |
🟢 Metrics 47 complexity
Metric Results Complexity 47
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[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces global cross-provenance linked edge calculation and delta sync capabilities, adding support for minimal delta updates (inserts and deletes) to Spanner. It updates the LinkedEdgeGenerator, AggregationDeleter, and AggregationOrchestrator to manage these global calculations in Phase 2, along with adding comprehensive unit tests. The review feedback highlights two key issues: a critical bug in _sync_edge_deltas where checking "provenance" in row on BigQuery Row objects incorrectly checks values instead of keys (causing Spanner deletions to fail), and a SQL injection/syntax risk in _active_imports_have_predicates that should be resolved by using parameterized queries instead of string formatting.
| row.get("provenance") | ||
| if hasattr(row, "get") | ||
| else ( | ||
| row["provenance"] | ||
| if "provenance" in row | ||
| else output_provenance | ||
| ), |
There was a problem hiding this comment.
In production, self.executor.client.query(...).result() returns google.cloud.bigquery.table.Row objects, which do not have a get method. Thus, the code falls back to the else block: row["provenance"] if "provenance" in row else output_provenance.
However, because Row behaves like a sequence/tuple, "provenance" in row checks if the string "provenance" is one of the values in the row, which evaluates to False. This causes the code to incorrectly fall back to output_provenance (e.g., generated/LinkedPlaces) instead of using the actual provenance of the row (e.g., generated/custom_import). As a result, the Spanner delete mutations will use the wrong primary key and silently fail to delete the stale/zombie edges.
This bug was missed because the unit test mocks the rows as dict objects (where hasattr(row, "get") is True).
To fix this, check "provenance" in row.keys() instead of "provenance" in row.
| row.get("provenance") | |
| if hasattr(row, "get") | |
| else ( | |
| row["provenance"] | |
| if "provenance" in row | |
| else output_provenance | |
| ), | |
| row.get("provenance") | |
| if hasattr(row, "get") | |
| else ( | |
| row["provenance"] | |
| if "provenance" in row.keys() | |
| else output_provenance | |
| ), |
| formatted_provenances = ", ".join( | ||
| [ | ||
| f"'{get_provenance_name(name, self.is_base_dc)}'" | ||
| for name in import_names | ||
| ] | ||
| ) | ||
| formatted_predicates = ", ".join([f"'{pred}'" for pred in predicates]) | ||
|
|
||
| query = f""" | ||
| SELECT 1 FROM Edge@{{FORCE_INDEX=EdgeByProvenance}} | ||
| WHERE provenance IN ({formatted_provenances}) | ||
| AND predicate IN ({formatted_predicates}) | ||
| LIMIT 1; | ||
| """ | ||
| try: | ||
| with self.spanner_database.snapshot() as snapshot: | ||
| results = list(snapshot.execute_sql(query)) | ||
| return len(results) > 0 |
There was a problem hiding this comment.
The _active_imports_have_predicates method constructs a SQL query using string formatting with import_names and predicates. Since import_names can be dynamically provided via CLI arguments or configuration, this introduces a risk of SQL syntax errors or SQL injection if any name contains single quotes.
Instead of manual string formatting, use parameterized queries with Spanner's UNNEST operator, which is safer, cleaner, and already used elsewhere in the codebase (e.g., in deleter.py).
query = """
SELECT 1 FROM Edge@{FORCE_INDEX=EdgeByProvenance}
WHERE provenance IN UNNEST(@provenances)
AND predicate IN UNNEST(@predicates)
LIMIT 1;
"""
provenances = [get_provenance_name(name, self.is_base_dc) for name in import_names]
params = {
"provenances": provenances,
"predicates": predicates,
}
param_types = {
"provenances": spanner.param_types.Array(spanner.param_types.STRING),
"predicates": spanner.param_types.Array(spanner.param_types.STRING),
}
try:
with self.spanner_database.snapshot() as snapshot:
results = list(
snapshot.execute_sql(
query, params=params, param_types=param_types
)
)
return len(results) > 01 participant
Summary
Enables global, cross-provenance transitive closures (
linkedContainedInPlace,linkedMemberOf, andlinkedMember) in post-processing aggregations, allowing custom hierarchies to link across dataset provenances in Data Commons Platform (DCP).Introduces
--enable_global_linked_edgesto control when this pipeline is active. WhenFalse(default), execution remains 100% legacy/scoped with zero behavioral changes.Key Changes
--enable_global_linked_edgesFlag: Controls whetherLINKED_EDGESexecutes as a Phase 2 global calculation or per-import in Phase 1 (default:False).generated/LinkedPlaces,generated/LinkedTopics, andgenerated/LinkedSVGs.EXCEPT DISTINCT) to isolate net changes:EXPORT DATA ... CLOUD_SPANNER.(subject_id, predicate, object_id, provenance)in 5,000-key batches.generated/{import_name}) on first run.EdgeByProvenanceindex withLIMIT 1to skip BigQuery calculation when active imports contain no schema triples (<15ms).Testing
python3 -m unittest aggregation/aggregation_test.py aggregation/orchestrator_test.py aggregation/deleter_test.py aggregation/validator_test.py