[DCP Ingestion] Fully multi threaded ingestion preprocessor

[gmechali]

Summary of Changes

• Parallel CSV Ingestion: Leverages ProcessPoolExecutor in DCP_BRIDGE mode to parallelize CSV imports across worker processes.
• Unified JSON-LD Directory Streaming: Workers share the master timestamped jsonld_dir, outputting all shards to a single GCS path for clean wildcard handshakes.
• Multi-Type Support: Introduced _create_importer_for_file factory to seamlessly handle OBSERVATIONS, EVENTS, and ENTITIES in parallel.
• Diagnostics & Node State Merging: Collects collision metrics, sample collision strings, and resolved place/entity nodes from worker processes into parent state.
• Pickle-Safe Exception Handling: Updated EntityResolutionError for safe multiprocessing pickling and diagnostic error reporting.

22M nodes, processed in 15 minutes on Cloud Run Job 8-core (down from 55 minutes), or 5mins 8 seconds on cloud batch 32core.

[codacy-production]

Not up to standards ⛔

🔴 Issues 7 medium · 3 minor

Alerts:
⚠ 10 issues (≤ 0 issues of at least minor severity)

Results:
10 new issues

Category	Results
UnusedCode	2 medium 1 minor
Documentation	1 minor
CodeStyle	1 minor
Complexity	5 medium

View in Codacy

🟢 Metrics 30 complexity

Metric	Results
Complexity	30

View in Codacy

_{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]

Code Review

This pull request introduces multiprocessing to parallelize CSV imports in DCP_BRIDGE mode using ProcessPoolExecutor, refactors observation shard writing and collision tracking to support chunked processing, and optimizes custom dimension serialization in variable_per_row_importer.py. The code review highlights several critical issues with the new multiprocessing implementation: subprocesses write to fragmented output directories due to independent timestamp generation, non-observation import types (such as events and entities) are silently skipped, and detailed collision tracking data is lost because it is not returned from the subprocesses and merged back into the main process.

[gmechali]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces parallel CSV importing using a process pool in DCP_BRIDGE mode, optimizes custom dimension serialization in VariablePerRowImporter, adds GCS upload retries, and improves observation collision tracking. Feedback on these changes highlights several critical issues: running imports in separate processes causes registered nodes (variables, event types, etc.) and cross-file collision detection to be lost in the parent process; the lack of context managers for Store objects in child processes leaks file descriptors; parallel processing of different data.csv files can lead to overwritten shards; and there is dead code in events_importer.py due to column renaming.

[gmechali]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces multiprocessing for importing CSV and MCF files in parallel under DCP_BRIDGE mode, optimizes custom dimension serialization in VariablePerRowImporter, and adds GCS upload retries and file-specific collision tracking in JsonLdStreamDb. While these changes improve performance and robustness, several critical issues were identified in the multiprocessing implementation. Specifically, custom properties created in worker processes are lost because they are not returned or merged into the parent process, and hardcoding the input directory path to the first store will cause failures when multiple input stores are used. Additionally, checking for MCF files using endswith instead of in violates repository guidelines for sharded files, and the signature of _write_observation_shard was unnecessarily refactored in a way that degrades type safety.