[DCP Ingestion] Namespace support in preprocessor#686
Conversation
Not up to standards ⛔🔴 Issues
|
| Category | Results |
|---|---|
| UnusedCode | 2 minor |
| Security | 3 medium |
| Complexity | 2 medium |
🟢 Metrics 44 complexity
Metric Results Complexity 44
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 parallel processing for CSV imports using a ProcessPoolExecutor in the Runner, along with changes to prevent file collisions by incorporating the file name into observation shard names. It also adds GCS upload retry logic, improves custom namespace handling across importers and validators, and optimizes custom dimension serialization. Feedback on these changes highlights a fragile design pattern in _write_observation_shard where the first parameter is overloaded to accept either a list or a tuple, and warns of a potential collision risk for node shards similar to the one resolved for observation shards.
| if file_name: | ||
| sanitized_stem = file_name.replace("/", "_").rsplit(".", 1)[0] | ||
| shard_name = f"observation-{sanitized_stem}-{shard_index:05d}.jsonld" | ||
| else: | ||
| shard_name = f"observation-{shard_index:05d}.jsonld" |
There was a problem hiding this comment.
While adding file_name to the observation shard name successfully prevents file collisions during parallel processing of multiple CSV files, a similar collision risk exists for node shards (which are written by MCF and Events CSV files).
Since _write_node_shard is not modified in this PR, please ensure that a follow-up change or a refactor is done to also pass and include the file_name in the node shard names (e.g., node-{sanitized_stem}-{shard_index:05d}.jsonld) to prevent them from overwriting each other when multiple MCF/Event files are processed in parallel.
| if isinstance(chunk_or_args, tuple): | ||
| chunk = chunk_or_args[0] | ||
| shard_index = chunk_or_args[1] | ||
| jsonld_dir_path = chunk_or_args[2] | ||
| ns_map = chunk_or_args[3] | ||
| prov_urls = chunk_or_args[4] | ||
| else: | ||
| chunk = chunk_or_args |
There was a problem hiding this comment.
Overloading the first parameter chunk_or_args to accept either a list of rows or a tuple of arguments is a fragile design pattern. If chunk is ever represented as a tuple of rows (e.g., from certain pandas/numpy operations or serialization), isinstance(chunk_or_args, tuple) will evaluate to True, leading to incorrect unpacking and silent failures or crashes.
Since _write_observation_shard is always called with individual arguments in this codebase, this tuple-unpacking branch appears to be dead/speculative code. It is highly recommended to remove this branch and keep the function signature clean and explicit.
chunk = chunk_or_args
1 participant
No description provided.