Fail node id enumeration fast on NULL node ids

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

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},

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)