[DCP Ingestion] Namespace support in preprocessor

simple/stats/data.py

@@ -28,6 +28,7 @@
 from stats import schema_constants as sc
 from stats.util import base64_decode_and_gunzip_json
 from stats.util import gzip_and_base64_encode_json
+from stats.util import has_namespace_prefix
 from stats.util import is_uri_or_namespace
 
 _PREDICATE_TYPE_OF = "typeOf"
@@ -181,6 +182,7 @@ def triples(self) -> list[Triple]:
 class Entity:
   entity_dcid: str
   entity_type: str
+  provenance_ids: set[str] = field(default_factory=set)
 
   def triples(self) -> list[Triple]:
     # Currently only 1 triple is generated but could be more in the future (e.g. name)
@@ -196,6 +198,7 @@ class Provenance:
   name: str
   url: str = ""
   properties: dict[str, str] = field(default_factory=dict)
+  provenance_id: str = ""
 
   def triples(self) -> list[Triple]:
     triples: list[Triple] = []
@@ -227,6 +230,7 @@ class Source:
   url: str = ""
   domain: str = field(init=False)
   properties: dict[str, str] = field(default_factory=dict)
+  provenance_id: str = ""
 
   def __post_init__(self):
     self.domain = urlparse(self.url).netloc
@@ -331,6 +335,7 @@ def _get_flattened_dataclass_field_names(cls) -> list[str]:
 class Property:
   dcid: str
   name: str
+  provenance_ids: set[str] = field(default_factory=set)
 
   def triples(self) -> list[Triple]:
     return [
@@ -546,7 +551,10 @@ def add_triple(self, triple: Triple) -> Self:
 
   def to_mcf(self) -> str:
     parts: list[str] = []
-    parts.append(f"Node: dcid:{self.id}")
+    if has_namespace_prefix(self.id):
+      parts.append(f"Node: {self.id}")
+    else:
+      parts.append(f"Node: dcid:{self.id}")
     parts.extend([f"{p}: {v}" for p, v in self.properties.items()])
     return "\n".join(parts)
 

simple/stats/db.py

@@ -230,7 +230,10 @@ class Db:
   def maybe_clear_before_import(self):
     pass
 
-  def insert_triples(self, triples: list[Triple], input_file: File = None):
+  def insert_triples(self,
+                     triples: list[Triple],
+                     input_file: File = None,
+                     provenance_dir: str = None):
     pass
 
   def insert_observations(self, observations_df: pd.DataFrame,
@@ -281,7 +284,10 @@ def __init__(self, db_params: dict) -> None:
     # dcid to node dict
     self.nodes: dict[str, McfNode] = {}
 
-  def insert_triples(self, triples: list[Triple], input_file: File = None):
+  def insert_triples(self,
+                     triples: list[Triple],
+                     input_file: File = None,
+                     provenance_dir: str = None):
     for triple in triples:
       self._add_triple(triple)
 
@@ -351,7 +357,10 @@ def __init__(self, config: dict) -> None:
   def maybe_clear_before_import(self):
     self.engine.clear_tables_and_indexes()
 
-  def insert_triples(self, triples: list[Triple], input_file: File = None):
+  def insert_triples(self,
+                     triples: list[Triple],
+                     input_file: File = None,
+                     provenance_dir: str = None):
     logging.info("Writing %s triples to [%s]", len(triples), self.engine)
     if triples:
       self.engine.executemany(_INSERT_TRIPLES_STATEMENT,
@@ -425,7 +434,10 @@ def maybe_clear_before_import(self):
     # Not applicable for Data Commons Platform.
     pass
 
-  def insert_triples(self, triples: list[Triple], input_file: File = None):
+  def insert_triples(self,
+                     triples: list[Triple],
+                     input_file: File = None,
+                     provenance_dir: str = None):
     """
     Convert triples to a jsonld graph and writes the graph to the Data Commons Platform instance.
     """

simple/stats/events_importer.py

@@ -20,13 +20,16 @@
 from stats.data import AggregationConfig
 from stats.data import Event
 from stats.data import filter_invalid_observation_values
+from stats.data import strip_namespace
 from stats.data import strip_namespace_series
 from stats.data import TimePeriod
 from stats.data import Triple
 from stats.db import Db
 from stats.importer import Importer
 from stats.nodes import Nodes
 from stats.reporter import FileImportReporter
+from stats.util import get_namespace_prefix_and_suffix
+from stats.util import has_namespace_prefix
 from util.filesystem import File
 
 from util import dc_client as dc
@@ -237,9 +240,9 @@ def _resolve_entities(self) -> None:
     pre_resolved_entities = {}
 
     def remove_pre_resolved(entity: str) -> bool:
-      if entity.startswith(constants.DCID_OVERRIDE_PREFIX):
-        pre_resolved_entities[entity] = entity[
-            len(constants.DCID_OVERRIDE_PREFIX):].strip()
+      if has_namespace_prefix(entity):
+        prefix, suffix = get_namespace_prefix_and_suffix(entity)
+        pre_resolved_entities[entity] = suffix.strip()
         return False
       return True
 
@@ -277,6 +280,19 @@ def remove_pre_resolved(entity: str) -> bool:
         unresolved=unresolved_list,
     )
 
+    prov_id = getattr(self, "provenance", "")
+    for dcid in df[constants.COLUMN_DCID].dropna().unique():
+      clean_dcid = strip_namespace(dcid)
+      if self.nodes.has_entity(clean_dcid):
+        if prov_id:
+          self.nodes.entities[clean_dcid].provenance_ids.add(prov_id)
+      elif prov_id:
+        self.nodes.entity_with_type(
+            clean_dcid,
+            self.entity_type or "Thing",
+            provenance_id=prov_id,
+        )
+
   def _resolve(self, entities: list[str]) -> dict[str, str]:
     lower_case_entity_name = self.entity_column_name.lower()
 

simple/stats/jsonld_stream_db.py

@@ -102,21 +102,13 @@ def _parse_numeric(val):
     return str(val)
 
 
-def _write_observation_shard(chunk_or_args,
+def _write_observation_shard(chunk,
                              shard_index: Optional[int] = None,
                              jsonld_dir_path: Optional[str] = None,
                              ns_map: Optional[dict[str, str]] = None,
                              prov_urls: Optional[dict[str, str]] = None,
                              track_hash_fn: Optional[Callable] = None,
                              file_name: str = ""):
-  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
 
   graph_list = []
   chunk_hashes = []
@@ -207,6 +199,9 @@ def _write_observation_shard(chunk_or_args,
   logging.info(f"Saved JSON-LD shard to {shard_name}")
 
 
+# TODO(gmechali): When parallelizing node shard exports (MCF and Events CSV files),
+# include the sanitized file_name in the shard name (e.g., node-{sanitized_stem}-{shard_index:05d}.jsonld)
+# to prevent shard filename collisions across parallel workers, similar to observation shards.
 def _write_node_shard(args):
   # TODO(gmechali): Get rid of this and keep only the "fast" mode.
   fast_export = os.getenv("FAST_NODE_EXPORT",
@@ -268,6 +263,8 @@ def _write_node_shard_fast(args):
   graph_list = sorted(list(subjects.values()), key=lambda x: x["@id"])
   compacted_jsonld = {"@context": ns_map, "@graph": graph_list}
 
+  # TODO(gmechali): When parallelizing node shard exports, include file_name in shard_name
+  # (e.g., f"node-{sanitized_stem}-{shard_index:05d}.jsonld") to prevent collisions across workers.
   shard_name = f"node-{shard_index:05d}.jsonld"
   with create_store(jsonld_dir_path) as store:
     output_dir = store.as_dir()
@@ -465,38 +462,43 @@ def _write_triples_to_disk(self, triples: list[Triple], import_name: str):
             (chunk, self.node_shard_index, import_temp_dir, self.ns_map))
         self.node_shard_index += 1
 
-  def insert_triples(self, triples: list[Triple], input_file: File = None):
+  def insert_triples(self,
+                     triples: list[Triple],
+                     input_file: File = None,
+                     provenance_dir: str = None):
     if not triples:
       return
 
-    if not input_file:
+    if not provenance_dir and input_file:
+      provenance_dir = self.config.import_name(input_file)
+
+    if not provenance_dir:
       with self.lock:
         self._triples["_global"].extend(triples)
       return
 
-    import_name = self.config.import_name(input_file)
-    self._init_import_export_dir(import_name)
-    self._write_triples_to_disk(triples, import_name)
+    self._init_import_export_dir(provenance_dir)
+    self._write_triples_to_disk(triples, provenance_dir)
 
   def commit(self):
     pass
 
   def commit_and_close(self):
-    # Add global triples to every processed import's triples
     global_triples = self._triples.pop("_global", [])
     if not self._processed_imports:
       self._processed_imports.add(self.import_name)
 
-    # Write global triples as node shards to the local temp directory for each import
-    for import_name in self._processed_imports:
-      import_temp_dir = os.path.join(self.temp_local_dir, import_name)
+    # Write any remaining untagged triples once to the primary import directory (no duplication across provenances)
+    if global_triples:
+      target_import = self.import_name if self.import_name in self._processed_imports else next(
+          iter(self._processed_imports))
+      import_temp_dir = os.path.join(self.temp_local_dir, target_import)
       os.makedirs(import_temp_dir, exist_ok=True)
-      if global_triples:
-        for i in range(0, len(global_triples), _CHUNK_SIZE):
-          chunk = global_triples[i:i + _CHUNK_SIZE]
-          _write_node_shard(
-              (chunk, self.node_shard_index, import_temp_dir, self.ns_map))
-          self.node_shard_index += 1
+      for i in range(0, len(global_triples), _CHUNK_SIZE):
+        chunk = global_triples[i:i + _CHUNK_SIZE]
+        _write_node_shard(
+            (chunk, self.node_shard_index, import_temp_dir, self.ns_map))
+        self.node_shard_index += 1
 
     has_local_files = any(os.scandir(self.temp_local_dir)) if os.path.exists(
         self.temp_local_dir) else False
@@ -591,7 +593,10 @@ def _upload_shards_local(self, temp_local_dir: str, files: list[str]):
 
     parent_dirs = set(os.path.dirname(f) for f in files if os.path.dirname(f))
     for d in sorted(parent_dirs):
-      target_store.open_dir(d)
+      try:
+        target_store.open_dir(d)
+      except Exception:
+        pass
 
     def _copy_single(rel_path: str):
       local_file_path = os.path.join(temp_local_dir, rel_path)

simple/stats/mcf_importer.py

@@ -105,7 +105,10 @@ def do_import(self) -> None:
 
         # Register all collected metadata nodes at the end
         if all_metadata_triples:
-          _register_metadata_nodes(all_metadata_triples, self.nodes)
+          prov_id = getattr(self, "provenance", "")
+          _register_metadata_nodes(all_metadata_triples,
+                                   self.nodes,
+                                   provenance_id=prov_id)
 
       self.reporter.report_success()
     except Exception as e:
@@ -161,7 +164,9 @@ def _to_triple(parser_triple: list[str], local2dcid: dict[str, str]) -> Triple:
   return Triple(resolved_subject, predicate, object_value=value)
 
 
-def _register_metadata_nodes(triples: list[Triple], nodes: Nodes) -> None:
+def _register_metadata_nodes(triples: list[Triple],
+                             nodes: Nodes,
+                             provenance_id: str = "") -> None:
   """Extracts and registers Provenance and Source nodes from parsed MCF triples.
 
   This helper is robust against:
@@ -191,8 +196,8 @@ def _register_metadata_nodes(triples: list[Triple], nodes: Nodes) -> None:
       val = triple.object_value or triple.object_id or ""
       subject_properties[sub_id]["name"] = _clean_literal(val)
     elif pred == "source":
-      # Map source to the source ID
-      subject_properties[sub_id]["source"] = triple.object_id
+      val = triple.object_id or triple.object_value or ""
+      subject_properties[sub_id]["source"] = _clean_literal(val)
 
   for sub_id, props in subject_properties.items():
     node_type = props.get("typeOf")
@@ -204,7 +209,8 @@ def _register_metadata_nodes(triples: list[Triple], nodes: Nodes) -> None:
     elif node_type == "Source":
       nodes.register_source(id=sub_id,
                             name=props.get("name", ""),
-                            url=props.get("url", ""))
+                            url=props.get("url", ""),
+                            provenance_id=provenance_id)
 
 
 def _clean_literal(val: str) -> str: