diff --git a/CHANGELOG.md b/CHANGELOG.md index 768b645a8..6d6774c4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## [Unreleased] +### Fixed + +- Fail node id enumeration fast when the source node table contains NULL node ids, instead of silently enumerating NULL + as its own node (https://github.com/Snapchat/GiGL/issues/288) + ## [0.3.1] - Jun 4, 2026 ### Fixed diff --git a/gigl/src/data_preprocessor/lib/enumerate/queries.py b/gigl/src/data_preprocessor/lib/enumerate/queries.py index 4dca9f2c4..16e1f46e4 100644 --- a/gigl/src/data_preprocessor/lib/enumerate/queries.py +++ b/gigl/src/data_preprocessor/lib/enumerate/queries.py @@ -3,8 +3,20 @@ UNIQUE_NODE_ENUMERATION_QUERY = """ WITH + source_node_ids AS ( + -- Fail fast if any node id is NULL. Otherwise, NULL would survive SELECT DISTINCT and be + -- silently enumerated as its own node, and edges referencing it would fail to map to any + -- enumerated node id downstream. See https://github.com/Snapchat/GiGL/issues/288. + SELECT + IF( + {bq_source_table_node_id_col_name} IS NULL, + ERROR("Found NULL node id in column `{bq_source_table_node_id_col_name}` of table `{bq_source_table_name}`; node ids must be non-NULL."), + {bq_source_table_node_id_col_name} + ) AS {original_node_id_field} + FROM `{bq_source_table_name}` + ), unique_nodes AS ( - SELECT DISTINCT {bq_source_table_node_id_col_name} as {original_node_id_field} FROM `{bq_source_table_name}` + SELECT DISTINCT {original_node_id_field} FROM source_node_ids ) SELECT {original_node_id_field}, diff --git a/tests/integration/pipeline/data_preprocessor/enumerator_test.py b/tests/integration/pipeline/data_preprocessor/enumerator_test.py index 0d06a9b1a..38a144c33 100644 --- a/tests/integration/pipeline/data_preprocessor/enumerator_test.py +++ b/tests/integration/pipeline/data_preprocessor/enumerator_test.py @@ -16,6 +16,7 @@ Enumerator, EnumeratorEdgeTypeMetadata, EnumeratorNodeTypeMetadata, + get_enumerated_node_id_map_bq_table_name, ) from gigl.src.data_preprocessor.lib.ingest.bigquery import ( BigqueryEdgeDataReference, @@ -599,6 +600,66 @@ def test_sharded_enumeration(self): edge_data_references=sharded_edge_references, ) + def test_enumeration_fails_on_null_node_ids(self) -> None: + """Enumeration should fail loudly if the source node table contains NULL node ids. + + See https://github.com/Snapchat/GiGL/issues/288: previously, a NULL node id would + survive SELECT DISTINCT, be silently assigned an enumerated int id, and edges + referencing it would not map to any enumerated node id downstream. + """ + null_node_data_reference = BigqueryNodeDataReference( + reference_uri=BqUtils.join_path( + get_resource_config().project, + get_resource_config().temp_assets_bq_dataset_name, + f"null_node_features_{self.__applied_task_identifier}", + ), + node_type=_PERSON_NODE_TYPE, + identifier=_PERSON_NODE_IDENTIFIER_FIELD, + ) + self.__bq_tables_to_cleanup_on_teardown.append( + null_node_data_reference.reference_uri + ) + # Note: the NULL node id is intentionally not in the first record, since + # __upload_records_to_bq infers the BQ schema from the first record's value types. + records_with_null_node_id: list[dict[str, Any]] = ( + _PERSON_NODE_FEATURE_RECORDS + + [ + { + _PERSON_NODE_IDENTIFIER_FIELD: None, + "height": 4.0, + "age": 4.0, + "weight": 4.0, + } + ] + ) + self.__upload_records_to_bq( + data_reference=null_node_data_reference, + records=records_with_null_node_id, + ) + + applied_task_identifier = AppliedTaskIdentifier( + f"{self.__applied_task_identifier}_null_node_id_enumeration" + ) + # Register the would-be output table for cleanup in case of a regression where + # the enumeration query unexpectedly succeeds. + self.__bq_tables_to_cleanup_on_teardown.append( + get_enumerated_node_id_map_bq_table_name( + applied_task_identifier=applied_task_identifier, + node_type=_PERSON_NODE_TYPE, + ) + ) + + # Enumerator.run() converts raised exceptions into sys.exit(...), so a failed + # enumeration surfaces as SystemExit. + with self.assertRaises(SystemExit) as ctx: + Enumerator().run( + applied_task_identifier=applied_task_identifier, + node_data_references=[null_node_data_reference], + edge_data_references=[], + gcp_project=get_resource_config().project, + ) + self.assertIn("NULL node id", str(ctx.exception)) + def tearDown(self) -> None: for table_name in self.__bq_tables_to_cleanup_on_teardown: self.__bq_utils.delete_bq_table_if_exist(bq_table_path=table_name)