From af256f6213f7b99a0349cafcc43864adfec1ee15 Mon Sep 17 00:00:00 2001 From: Christie Ellks Date: Tue, 23 Jun 2026 02:52:19 -0700 Subject: [PATCH] Align observationProperties to JSON-LD reference objects and fix custom namespace lookups 1. Preprocessor Alignment: - Updated 'jsonld_stream_db.py' to generate proper JSON-LD reference objects ('{"@id": "..."}') for 'observationProperties' keys rather than plain strings, ensuring semantic correctness and full compliance with JSON-LD specs. 2. Robust Namespace Resolution in Java Pipeline: - Discovered and fixed a critical bug in 'McfUtil.stripNamespace' where property lookups would fail for custom/third-party namespaces (e.g. 'custom:destinationCountry') because it was hardcoded to only strip 'dcid:', 'schema:', and 'dcs:'. - Updated the implementation to dynamically strip any alphanumeric namespace prefix before a colon, and also handle full expanded URL prefixes, ensuring DCP's open-source flexibility. 3. Testing Coverage: - Added 'testNestedObservationProperties_StringList' in 'JsonLdParserTest.java' to verify parsing of string-list fallback formats. - Added 'testExtractTimeSeries_JsonLd_MultiEntity_Integration' in 'GraphReaderTest.java' to verify end-to-end parsing and reading of JSON-LD with custom namespaces, successfully validating the fix. - Added corresponding assertions in 'jsonld_stream_db_test.py' to prevent regressions. TAG=agy CONV=09581de6-99cc-4213-8266-f3dbf7e70a52 --- .../ingestion/util/GraphReaderTest.java | 47 ++++++++++++++++ simple/stats/jsonld_stream_db.py | 4 +- simple/tests/stats/jsonld_stream_db_test.py | 8 +++ .../java/org/datacommons/util/McfUtil.java | 19 +++++-- .../datacommons/util/JsonLdParserTest.java | 56 +++++++++++++++++++ 5 files changed, 127 insertions(+), 7 deletions(-) diff --git a/pipeline/util/src/test/java/org/datacommons/ingestion/util/GraphReaderTest.java b/pipeline/util/src/test/java/org/datacommons/ingestion/util/GraphReaderTest.java index 2c0ac68a2..1b5a16688 100644 --- a/pipeline/util/src/test/java/org/datacommons/ingestion/util/GraphReaderTest.java +++ b/pipeline/util/src/test/java/org/datacommons/ingestion/util/GraphReaderTest.java @@ -604,4 +604,51 @@ public void testExtractTimeSeries_MissingEntity() { // This should throw RuntimeException because sourceCountry is missing GraphReader.extractTimeSeries("nodeId", pv, "test_import", true); } + + @Test + public void testExtractTimeSeries_JsonLd_MultiEntity_Integration() throws Exception { + String jsonLd = + "{\n" + + " \"@context\": {\n" + + " \"custom\": \"https://customsource.org/schema/\",\n" + + " \"dcid\": \"https://datacommons.org/browser/\"\n" + + " },\n" + + " \"@graph\": [\n" + + " {\n" + + " \"@id\": \"dcid:TestObs\",\n" + + " \"@type\": \"dcid:StatVarObservation\",\n" + + " \"dcid:variableMeasured\": {\n" + + " \"@id\": \"custom:FinancialTrade\",\n" + + " \"dcid:observationProperties\": [\n" + + " \"custom:destinationCountry\",\n" + + " \"custom:sourceCountry\"\n" + + " ]\n" + + " },\n" + + " \"dcid:observationDate\": \"2024\",\n" + + " \"dcid:value\": 100,\n" + + " \"custom:sourceCountry\": { \"@id\": \"dcid:country/FRA\" },\n" + + " \"custom:destinationCountry\": { \"@id\": \"dcid:country/USA\" }\n" + + " }\n" + + " ]\n" + + "}"; + + java.io.InputStream in = + new java.io.ByteArrayInputStream(jsonLd.getBytes(java.nio.charset.StandardCharsets.UTF_8)); + McfGraph graph = org.datacommons.util.parser.jsonld.JsonLdParser.parse(in); + + PropertyValues pv = graph.getNodesMap().get("TestObs"); + org.junit.Assert.assertNotNull(pv); + + TimeSeries expected = + TimeSeries.builder() + .entity1("country/USA") + .extraEntities(List.of("country/FRA")) + .variableMeasured("FinancialTrade") + .importName("test_import") + .isBaseDc(true) + .build(); + + TimeSeries actual = GraphReader.extractTimeSeries("TestObs", pv, "test_import", true); + assertEquals(expected, actual); + } } diff --git a/simple/stats/jsonld_stream_db.py b/simple/stats/jsonld_stream_db.py index 54908dc83..d46a7f723 100644 --- a/simple/stats/jsonld_stream_db.py +++ b/simple/stats/jsonld_stream_db.py @@ -91,9 +91,7 @@ def _write_observation_shard(args): props_dict = json.loads(props) if isinstance(props_dict, dict): prop_keys = [ - f"dcid:{k}" if not k.startswith( - ("dcid:", "http://", "https://")) else k - for k in props_dict.keys() + _uri_ref(k) for k in props_dict.keys() ] if prop_keys and var_obj: var_obj["dcid:observationProperties"] = prop_keys diff --git a/simple/tests/stats/jsonld_stream_db_test.py b/simple/tests/stats/jsonld_stream_db_test.py index e13c5ee06..d5aea333e 100644 --- a/simple/tests/stats/jsonld_stream_db_test.py +++ b/simple/tests/stats/jsonld_stream_db_test.py @@ -287,6 +287,14 @@ def test_observation_parsing_edge_cases(self): o for o in graph if o["dcid:observationAbout"]["@id"] == "dcid:country/ALB" ][0] + self.assertEqual(obs1["dcid:variableMeasured"], { + "@id": "dcid:v1", + "dcid:observationProperties": [ + {"@id": "dcid:customIntProp"}, + {"@id": "dcid:customStrProp"}, + {"@id": "http://schema.org/url"} + ] + }) self.assertEqual(obs1["dcid:value"], 99) self.assertEqual(obs1["dcid:observationDate"], 2026) self.assertEqual(obs1["dcid:scalingFactor"], 100) diff --git a/util/src/main/java/org/datacommons/util/McfUtil.java b/util/src/main/java/org/datacommons/util/McfUtil.java index 23cd56d11..589821965 100644 --- a/util/src/main/java/org/datacommons/util/McfUtil.java +++ b/util/src/main/java/org/datacommons/util/McfUtil.java @@ -162,10 +162,21 @@ public static Mcf.McfGraph mergeGraphs(List graphs) throws Asserti } public static String stripNamespace(String val) { - if (val.startsWith(Vocabulary.DCID_PREFIX) - || val.startsWith(Vocabulary.SCHEMA_ORG_PREFIX) - || val.startsWith(Vocabulary.DC_SCHEMA_PREFIX)) { - return val.substring(val.indexOf(Vocabulary.REFERENCE_DELIMITER) + 1); + if (val == null) { + return null; + } + if (val.startsWith("http://") || val.startsWith("https://")) { + int lastSlash = val.lastIndexOf('/'); + int lastHash = val.lastIndexOf('#'); + int idx = Math.max(lastSlash, lastHash); + if (idx > 0 && idx < val.length() - 1) { + return val.substring(idx + 1); + } + return val; + } + int colonIdx = val.indexOf(Vocabulary.REFERENCE_DELIMITER); + if (colonIdx > 0) { + return val.substring(colonIdx + 1); } return val; } diff --git a/util/src/test/java/org/datacommons/util/JsonLdParserTest.java b/util/src/test/java/org/datacommons/util/JsonLdParserTest.java index 097382108..204371e6b 100644 --- a/util/src/test/java/org/datacommons/util/JsonLdParserTest.java +++ b/util/src/test/java/org/datacommons/util/JsonLdParserTest.java @@ -150,4 +150,60 @@ public void testNestedObservationProperties() throws Exception { assertTrue(foundDest); assertTrue(foundSource); } + + @Test + public void testNestedObservationProperties_StringList() throws Exception { + String jsonLd = + "{\n" + + " \"@context\": {\n" + + " \"custom\": \"https://customsource.org/schema/\",\n" + + " \"dcid\": \"https://datacommons.org/browser/\"\n" + + " },\n" + + " \"@graph\": [\n" + + " {\n" + + " \"@id\": \"dcid:TestObs\",\n" + + " \"@type\": \"dcid:StatVarObservation\",\n" + + " \"dcid:variableMeasured\": {\n" + + " \"@id\": \"custom:FinancialTrade\",\n" + + " \"dcid:observationProperties\": [\n" + + " \"custom:destinationCountry\",\n" + + " \"custom:sourceCountry\"\n" + + " ]\n" + + " },\n" + + " \"dcid:observationDate\": \"2024\",\n" + + " \"dcid:value\": 100,\n" + + " \"custom:sourceCountry\": { \"@id\": \"dcid:country/FRA\" },\n" + + " \"custom:destinationCountry\": { \"@id\": \"dcid:country/USA\" }\n" + + " }\n" + + " ]\n" + + "}"; + + InputStream in = new java.io.ByteArrayInputStream(jsonLd.getBytes(StandardCharsets.UTF_8)); + McfGraph graph = JsonLdParser.parse(in); + + assertNotNull(graph); + assertTrue(graph.getNodesMap().containsKey("TestObs")); + McfGraph.PropertyValues node = graph.getNodesMap().get("TestObs"); + + assertTrue(node.containsPvs("variableMeasured")); + assertTrue(node.containsPvs("observationProperties")); + java.util.List values = + node.getPvsMap().get("observationProperties").getTypedValuesList(); + assertTrue(values.size() == 2); + + boolean foundDest = false; + boolean foundSource = false; + for (McfGraph.TypedValue tv : values) { + if ("custom:destinationCountry".equals(tv.getValue()) + || "https://customsource.org/schema/destinationCountry".equals(tv.getValue())) { + foundDest = true; + } + if ("custom:sourceCountry".equals(tv.getValue()) + || "https://customsource.org/schema/sourceCountry".equals(tv.getValue())) { + foundSource = true; + } + } + assertTrue(foundDest); + assertTrue(foundSource); + } }