scope uniqueness constraints to specific objects

backend/infrahub/computed_attribute/tasks.py

@@ -21,6 +21,7 @@
 from infrahub.git.repository import get_initialized_repo
 from infrahub.trigger.models import TriggerSetupReport, TriggerType
 from infrahub.trigger.setup import setup_triggers, setup_triggers_specific
+from infrahub.utilities.chunks import chunked
 from infrahub.workers.dependencies import get_client, get_component, get_database, get_workflow
 from infrahub.workflows.catalogue import (
     COMPUTED_ATTRIBUTE_PROCESS_JINJA2,
@@ -54,10 +55,6 @@
     from infrahub.graphql.analyzer import GraphQLQueryReport
 
 
-def _chunk_ids(ids: list[str], chunk_size: int) -> list[list[str]]:
-    return [ids[i : i + chunk_size] for i in range(0, len(ids), chunk_size)]
-
-
 async def _reconcile_python_computed_attribute_automations(db: InfrahubDatabase) -> None:
     """Reconcile the node-input (data-path) automations against the current schema.
 
@@ -261,7 +258,7 @@ async def trigger_update_python_computed_attributes(
         return
 
     chunk_size = get_submission_chunk_size()
-    for chunk in _chunk_ids(object_ids, chunk_size):
+    for chunk in chunked(object_ids, chunk_size):
         await get_workflow().submit_workflow(
             workflow=COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM,
             context=context,
@@ -658,7 +655,7 @@ async def query_transform_targets(
 
     chunk_size = get_submission_chunk_size()
     for (kind, attribute_name), batch_object_ids in batches.items():
-        for chunk in _chunk_ids(batch_object_ids, chunk_size):
+        for chunk in chunked(batch_object_ids, chunk_size):
             await get_workflow().submit_workflow(
                 workflow=COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM,
                 context=context,

backend/infrahub/core/branch/tasks.py

@@ -35,7 +35,8 @@
 from infrahub.core.migrations.runner import MigrationRunner
 from infrahub.core.schema.update_coordinator import MigrationExecutor, SchemaUpdateCoordinator
 from infrahub.core.timestamp import Timestamp
-from infrahub.core.validators.determiner import ConstraintValidatorDeterminer
+from infrahub.core.validators.constraint_merge import build_constraint_info_merger
+from infrahub.core.validators.determiner import build_constraint_validator_determiner
 from infrahub.core.validators.models.validate_migration import SchemaValidateMigrationData
 from infrahub.core.validators.tasks import schema_validate_migrations
 from infrahub.dependencies.registry import get_component_registry
@@ -66,6 +67,7 @@
 if TYPE_CHECKING:
     from logging import Logger, LoggerAdapter
 
+    from infrahub.core.models import SchemaUpdateConstraintInfo
     from infrahub.database import InfrahubDatabase
 
 
@@ -175,14 +177,19 @@ async def rebase_branch(branch: str, context: InfrahubContext, send_events: bool
         )
 
         candidate_schema = schema_analyzer.get_candidate_schema()
-        determiner = ConstraintValidatorDeterminer(schema_branch=candidate_schema)
-        constraints = await determiner.get_constraints(node_diffs=node_diff_field_summaries)
+        determiner = build_constraint_validator_determiner(
+            db=db, branch=user_branch, schema_branch=candidate_schema, at=rebase_at
+        )
+        data_diff_constraints = await determiner.get_constraints(node_diffs=node_diff_field_summaries)
 
         # If there are some changes related to the schema between this branch and main, we need to
         #  - Run all the validations to ensure everything is correct before rebasing the branch
         #  - Run all the migrations after the rebase
+        schema_diff_constraints: list[SchemaUpdateConstraintInfo] = []
         if user_branch.has_schema_changes:
-            constraints += await schema_analyzer.calculate_validations(target_schema=candidate_schema)
+            schema_diff_constraints = await schema_analyzer.calculate_validations(target_schema=candidate_schema)
+        merger = build_constraint_info_merger(schema_branch=candidate_schema)
+        constraints = merger.merge(data_diff_constraints, schema_diff_constraints)
         if constraints:
             responses = await schema_validate_migrations(
                 message=SchemaValidateMigrationData(

backend/infrahub/core/diff/model/path.py

@@ -102,9 +102,41 @@ def __str__(self) -> str:
 
 @dataclass
 class NodeDiffFieldSummary:
+    """Which fields changed for a kind, and which nodes changed each one.
+
+    The per-field maps keep the correlation between a changed field and the nodes that changed it,
+    so a consumer can scope work to the nodes touching a specific field rather than every changed
+    node of the kind. ``attribute_names``/``relationship_names``/``node_uuids`` are derived views.
+    """
+
     kind: str
-    attribute_names: set[str] = field(default_factory=set)
-    relationship_names: set[str] = field(default_factory=set)
+    attribute_node_uuids: dict[str, set[str]] = field(default_factory=dict)
+    relationship_node_uuids: dict[str, set[str]] = field(default_factory=dict)
+
+    def add_attribute_node_uuid(self, name: str, node_uuid: str) -> None:
+        """Record that `node_uuid` is a node of this kind whose attribute `name` changed."""
+        self.attribute_node_uuids.setdefault(name, set()).add(node_uuid)
+
+    def add_relationship_node_uuid(self, name: str, node_uuid: str) -> None:
+        """Record that `node_uuid` is a node of this kind whose relationship `name` changed."""
+        self.relationship_node_uuids.setdefault(name, set()).add(node_uuid)
+
+    @property
+    def attribute_names(self) -> set[str]:
+        return set(self.attribute_node_uuids)
+
+    @property
+    def relationship_names(self) -> set[str]:
+        return set(self.relationship_node_uuids)
+
+    @property
+    def node_uuids(self) -> set[str]:
+        uuids: set[str] = set()
+        for field_uuids in self.attribute_node_uuids.values():
+            uuids |= field_uuids
+        for field_uuids in self.relationship_node_uuids.values():
+            uuids |= field_uuids
+        return uuids
 
 
 @dataclass

backend/infrahub/core/diff/query/field_summary.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from typing import Any
 
 from infrahub.core.constants import DiffAction
@@ -7,6 +8,14 @@
 from ..model.path import NodeDiffFieldSummary, TrackingId
 
 
+@dataclass
+class FieldNodeUuidsRow:
+    """One changed field of a kind and the uuids of the nodes that changed it, as projected by the query."""
+
+    name: str
+    node_uuids: list[str]
+
+
 class EnrichedDiffNodeFieldSummaryQuery(Query):
     """Get node kind and names of all altered attributes and relationships for each kind."""
 
@@ -47,30 +56,43 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:  # noqa
             OPTIONAL MATCH (n:DiffNode {kind: kind})-[:DIFF_HAS_ATTRIBUTE]->(a:DiffAttribute)
             WHERE n.action <> $unchanged_str
             AND a.action <> $unchanged_str
-            WITH DISTINCT a.name AS attr_name
-            RETURN collect(attr_name) AS attr_names
+            WITH a.name AS attr_name, collect(DISTINCT n.uuid) AS attr_node_uuids
+            WHERE attr_name IS NOT NULL
+            RETURN collect({name: attr_name, node_uuids: attr_node_uuids}) AS attr_name_uuids
         }
-        WITH kind, attr_names
+        WITH kind, attr_name_uuids
         CALL (kind) {
             OPTIONAL MATCH (n:DiffNode {kind: kind})-[:DIFF_HAS_RELATIONSHIP]->(r:DiffRelationship)
             WHERE n.action <> $unchanged_str
             AND r.action <> $unchanged_str
-            WITH DISTINCT r.name AS rel_name
-            RETURN collect(rel_name) AS rel_names
+            WITH r.name AS rel_name, collect(DISTINCT n.uuid) AS rel_node_uuids
+            WHERE rel_name IS NOT NULL
+            RETURN collect({name: rel_name, node_uuids: rel_node_uuids}) AS rel_name_uuids
         }
         """
         self.add_to_query(query=query)
         self.order_by = ["kind"]
-        self.return_labels = ["kind", "attr_names", "rel_names"]
+        self.return_labels = ["kind", "attr_name_uuids", "rel_name_uuids"]
 
     async def get_field_summaries(self) -> list[NodeDiffFieldSummary]:
         field_summaries = []
         for result in self.get_results():
             kind = result.get_as_type(label="kind", return_type=str)
-            attr_names = result.get_as_type(label="attr_names", return_type=list[str])
-            rel_names = result.get_as_type(label="rel_names", return_type=list[str])
-            if attr_names or rel_names:
+            attribute_node_uuids = self._to_field_uuids(
+                result.get_as_list_of_type(label="attr_name_uuids", return_type=FieldNodeUuidsRow)
+            )
+            relationship_node_uuids = self._to_field_uuids(
+                result.get_as_list_of_type(label="rel_name_uuids", return_type=FieldNodeUuidsRow)
+            )
+            if attribute_node_uuids or relationship_node_uuids:
                 field_summaries.append(
-                    NodeDiffFieldSummary(kind=kind, attribute_names=set(attr_names), relationship_names=set(rel_names))
+                    NodeDiffFieldSummary(
+                        kind=kind,
+                        attribute_node_uuids=attribute_node_uuids,
+                        relationship_node_uuids=relationship_node_uuids,
+                    )
                 )
         return field_summaries
+
+    def _to_field_uuids(self, rows: list[FieldNodeUuidsRow]) -> dict[str, set[str]]:
+        return {row.name: set(row.node_uuids) for row in rows}

backend/infrahub/core/merge/builder.py

@@ -9,6 +9,7 @@
 from infrahub.core.diff.repository.repository import DiffRepository
 from infrahub.core.registry import registry
 from infrahub.core.schema.update_coordinator import SchemaUpdateCoordinator
+from infrahub.core.validators.tasks import schema_validate_migrations
 from infrahub.dependencies.registry import get_component_registry
 from infrahub.workers.dependencies import get_cache, get_event_service, get_workflow
 
@@ -58,7 +59,12 @@ async def build_branch_merge_orchestrator(
         diff_repository=diff_repository,
         schema_manager=registry.schema,
     )
-    constraint_validator = MergeConstraintValidator(db=db, branch=source_branch, diff_repository=diff_repository)
+    constraint_validator = MergeConstraintValidator(
+        db=db,
+        branch=source_branch,
+        diff_repository=diff_repository,
+        migration_validator=schema_validate_migrations,
+    )
     graph_merger = GraphMerger(
         db=db,
         source_branch=source_branch,

backend/infrahub/core/merge/constraints.py

@@ -5,11 +5,13 @@
 
 from infrahub.core.diff.model.diff import SchemaConflict
 from infrahub.core.diff.model.path import BranchTrackingId
-from infrahub.core.validators.determiner import ConstraintValidatorDeterminer
+from infrahub.core.validators.constraint_merge import build_constraint_info_merger
+from infrahub.core.validators.determiner import build_constraint_validator_determiner
 from infrahub.core.validators.models.validate_migration import SchemaValidateMigrationData
-from infrahub.core.validators.tasks import schema_validate_migrations
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
     from infrahub.core.branch import Branch
     from infrahub.core.diff.repository.repository import DiffRepository
     from infrahub.core.models import SchemaUpdateConstraintInfo
@@ -18,6 +20,8 @@
     from infrahub.core.validators.models.validate_migration import SchemaValidatorPathResponseData
     from infrahub.database import InfrahubDatabase
 
+    MigrationValidator = Callable[[SchemaValidateMigrationData], Awaitable[list[SchemaValidatorPathResponseData]]]
+
 
 @dataclass
 class MergeConstraintValidationResult:
@@ -95,27 +99,35 @@ class MergeConstraintValidator:
     during the merge, so its pre-resolution cross-branch state is not a real violation.
     """
 
-    def __init__(self, db: InfrahubDatabase, branch: Branch, diff_repository: DiffRepository) -> None:
+    def __init__(
+        self,
+        db: InfrahubDatabase,
+        branch: Branch,
+        diff_repository: DiffRepository,
+        migration_validator: MigrationValidator,
+    ) -> None:
         self.db = db
         self.branch = branch
         self.diff_repository = diff_repository
+        self.migration_validator = migration_validator
 
     async def validate(
         self, candidate_schema: SchemaBranch, schema_diff_constraints: list[SchemaUpdateConstraintInfo]
     ) -> MergeConstraintValidationResult:
-        determiner = ConstraintValidatorDeterminer(schema_branch=candidate_schema)
+        determiner = build_constraint_validator_determiner(
+            db=self.db, branch=self.branch, schema_branch=candidate_schema
+        )
         node_field_summaries = await self.diff_repository.get_node_field_summaries(
             diff_branch_name=self.branch.name, tracking_id=BranchTrackingId(name=self.branch.name)
         )
         data_diff_constraints = await determiner.get_constraints(node_diffs=node_field_summaries)
-        constraints = set(data_diff_constraints + schema_diff_constraints)
+        merger = build_constraint_info_merger(schema_branch=candidate_schema)
+        constraints = merger.merge(data_diff_constraints, schema_diff_constraints)
         if not constraints:
             return MergeConstraintValidationResult()
 
-        responses = await schema_validate_migrations(
-            message=SchemaValidateMigrationData(
-                branch=self.branch, schema_branch=candidate_schema, constraints=list(constraints)
-            )
+        responses = await self.migration_validator(
+            SchemaValidateMigrationData(branch=self.branch, schema_branch=candidate_schema, constraints=constraints)
         )
         conflicted_fields = await gather_conflicted_fields(
             diff_repository=self.diff_repository, branch_name=self.branch.name

backend/infrahub/core/merge/recompute_coalescing.py

@@ -9,6 +9,7 @@
 from infrahub.events.limits import get_submission_chunk_size
 from infrahub.hfid.scoping import derive_hfid_targets
 from infrahub.log import get_logger
+from infrahub.utilities.chunks import chunked
 from infrahub.workflows.catalogue import (
     COMPUTED_ATTRIBUTE_PROCESS_JINJA2,
     DISPLAY_LABELS_PROCESS_JINJA2,
@@ -173,12 +174,6 @@ class CoalescedSubmission:
     node_ids: tuple[str, ...]
 
 
-def _chunk(ids: tuple[str, ...], size: int) -> Iterator[tuple[str, ...]]:
-    """Yield ``ids`` in contiguous slices of at most ``size``."""
-    for start in range(0, len(ids), size):
-        yield ids[start : start + size]
-
-
 class CoalescedRecomputeBuilder:
     """Derive the deduplicated recompute for a merge or rebase change set from one schema branch."""
 
@@ -339,7 +334,7 @@ def plan(coalesced: CoalescedRecompute) -> list[CoalescedSubmission]:
             )
             for target in coalesced.targets
             for lookup in target.reader_lookups
-            for chunk in _chunk(tuple(sorted(lookup.source_node_ids)), chunk_size)
+            for chunk in chunked(tuple(sorted(lookup.source_node_ids)), chunk_size)
         ]
         return sorted(
             submissions,

backend/infrahub/core/migrations/__init__.py

@@ -1,3 +1,5 @@
+from infrahub.core.validators.enum import ConstraintIdentifier
+
 from .schema.attribute_kind_update import AttributeKindUpdateMigration
 from .schema.attribute_name_update import AttributeNameUpdateMigration
 from .schema.attribute_supports_generated_schema import AttributeSupportsGeneratedSchemaMigration
@@ -18,7 +20,7 @@
     "node.name.update": NodeKindUpdateMigration,
     "node.namespace.update": NodeKindUpdateMigration,
     "node.relationship.remove": NodeRelationshipRemoveMigration,
-    "node.uniqueness_constraints.update": NodeUniquenessConstraintsUpdateMigration,
+    ConstraintIdentifier.NODE_UNIQUENESS_CONSTRAINTS_UPDATE.value: NodeUniquenessConstraintsUpdateMigration,
     "attribute.name.update": AttributeNameUpdateMigration,
     "attribute.branch.update": None,
     "attribute.kind.update": AttributeKindUpdateMigration,

backend/infrahub/core/migrations/graph/m018_uniqueness_nulls.py

@@ -56,7 +56,9 @@ async def validate_nulls_in_uniqueness_constraints(db: InfrahubDatabase) -> Migr
         if not includes_optional_attr:
             continue
 
-        non_unique_nodes = await uniqueness_checker.check_one_schema(schema=schema)
+        non_unique_nodes = await uniqueness_checker.check_one_schema(
+            schema=schema, branch=default_branch, schema_branch=schema_branch
+        )
         if non_unique_nodes:
             non_unique_nodes_by_kind[schema_kind] = non_unique_nodes
 

backend/infrahub/core/migrations/schema/node_uniqueness_constraints_update.py

@@ -6,6 +6,7 @@
 from infrahub.core.path import SchemaPath
 from infrahub.core.schema.generic_schema import GenericSchema
 from infrahub.core.schema.node_schema import NodeSchema
+from infrahub.core.validators.enum import ConstraintIdentifier
 
 from ..query import MigrationBaseQuery  # noqa: TC001
 from ..shared import AttributeSchemaMigration, MigrationInput, MigrationResult, SchemaMigration
@@ -19,7 +20,7 @@
 
 
 class NodeUniquenessConstraintsUpdateMigration(SchemaMigration):
-    name: str = "node.uniqueness_constraints.update"
+    name: str = ConstraintIdentifier.NODE_UNIQUENESS_CONSTRAINTS_UPDATE.value
     queries: Sequence[type[MigrationBaseQuery]] = []
 
     async def execute(

backend/infrahub/core/models.py

@@ -156,6 +156,11 @@ class SchemaUpdateConstraintInfo(BaseModel):
     model_config = ConfigDict(extra="forbid")
     path: SchemaPath
     constraint_name: str
+    node_uuids: list[str] | None = None
+    """When set, restrict validation to these nodes; None means validate the full population.
+
+    node_uuids=None indicates that all possible matching objects must be checked
+    """
 
     @property
     def routing_key(self) -> str: