Merge pull request #5967 from opsmill/bkr-merge-stable-release1.2-20250307

Merge stable  to release1.2 (2025-03-07)

Author: ajtmccarty

.vale/styles/spelling-exceptions.txt

@@ -109,6 +109,7 @@ regenerate_host_artifact
 repo
 REST
 resources
+schema's
 schema_mapping
 sdk
 subcommand

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

@@ -21,7 +21,8 @@
 from .m017_add_core_profile import Migration017
 from .m018_uniqueness_nulls import Migration018
 from .m019_restore_rels_to_time import Migration019
-from .m020_add_generate_template_attr import Migration020
+from .m020_duplicate_edges import Migration020
+from .m021_add_generate_template_attr import Migration021
 
 if TYPE_CHECKING:
     from infrahub.core.root import Root
@@ -49,6 +50,7 @@
     Migration018,
     Migration019,
     Migration020,
+    Migration021,
 ]
 
 

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

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Sequence
+
+from infrahub.core.constants.database import DatabaseEdgeType
+from infrahub.core.migrations.shared import GraphMigration, MigrationResult
+from infrahub.log import get_logger
+
+from ...query import Query, QueryType
+
+if TYPE_CHECKING:
+    from infrahub.database import InfrahubDatabase
+
+log = get_logger()
+
+
+class DeleteDuplicateHasValueEdgesQuery(Query):
+    name = "delete_duplicate_has_value_edges"
+    type = QueryType.WRITE
+    insert_return = False
+
+    async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> None:  # noqa: ARG002
+        query = """
+// -------------------
+// find Attribute nodes with multiple identical edges to AttributeValue nodes with the same value
+// -------------------
+MATCH (a:Attribute)-[e:HAS_VALUE]->(av:AttributeValue)
+WITH a, e.branch AS branch, e.branch_level AS branch_level, e.status AS status, e.from AS from, e.to AS to,
+    av.value AS attr_val, av.is_default AS attr_default, COUNT(*) AS num_duplicate_edges
+WHERE num_duplicate_edges > 1
+// -------------------
+// get the the one AttributeValue we want to use
+// -------------------
+WITH DISTINCT a, branch, branch_level, status, from, to, attr_val, attr_default
+WITH attr_val, attr_default, collect([a, branch, branch_level, status, from, to]) AS details_list
+CALL {
+    WITH attr_val, attr_default
+    MATCH (av:AttributeValue {value: attr_val, is_default: attr_default})
+    RETURN av AS the_one_av
+    ORDER by %(id_func)s(av) ASC
+    LIMIT 1
+}
+UNWIND details_list AS details_item
+WITH attr_val, attr_default, the_one_av,
+    details_item[0] AS a, details_item[1] AS branch, details_item[2] AS branch_level,
+    details_item[3] AS status, details_item[4] AS from, details_item[5] AS to
+// -------------------
+// get/create the one edge to keep
+// -------------------
+CREATE (a)-[fresh_e:HAS_VALUE {branch: branch, branch_level: branch_level, status: status, from: from}]->(the_one_av)
+SET fresh_e.to = to
+WITH a, branch, status, from, to, attr_val, attr_default, %(id_func)s(fresh_e) AS e_id_to_keep
+// -------------------
+// get the identical edges for a given set of Attribute node, edge properties, AttributeValue.value
+// -------------------
+CALL {
+    // -------------------
+    // delete the duplicate edges a given set of Attribute node, edge properties, AttributeValue.value
+    // -------------------
+    WITH a, branch, status, from, to, attr_val, attr_default, e_id_to_keep
+    MATCH (a)-[e:HAS_VALUE]->(av:AttributeValue {value: attr_val, is_default: attr_default})
+    WHERE %(id_func)s(e) <> e_id_to_keep
+    AND e.branch = branch AND e.status = status AND e.from = from
+    AND (e.to = to OR (e.to IS NULL AND to IS NULL))
+    DELETE e
+}
+// -------------------
+// delete any orphaned AttributeValue nodes
+// -------------------
+WITH NULL AS nothing
+LIMIT 1
+MATCH (orphaned_av:AttributeValue)
+WHERE NOT exists((orphaned_av)-[]-())
+DELETE orphaned_av
+        """ % {"id_func": db.get_id_function_name()}
+        self.add_to_query(query)
+
+
+class DeleteDuplicateBooleanEdgesQuery(Query):
+    name = "delete_duplicate_booleans_edges"
+    type = QueryType.WRITE
+    insert_return = False
+    edge_type: DatabaseEdgeType | None = None
+
+    async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> None:  # noqa: ARG002
+        if not self.edge_type:
+            raise RuntimeError("edge_type is required for this query")
+        query = """
+// -------------------
+// find Attribute nodes with multiple identical edges to Boolean nodes
+// -------------------
+MATCH (a:Attribute)-[e:%(edge_type)s]->(b)
+WITH a, e.branch AS branch, e.branch_level AS branch_level, e.status AS status, e.from AS from, e.to AS to, b, COUNT(*) AS num_duplicate_edges
+WHERE num_duplicate_edges > 1
+// -------------------
+// get the identical edges for a given set of Attribute node, edge properties, Boolean
+// -------------------
+WITH DISTINCT a, branch, branch_level, status, from, to, b
+CREATE (a)-[fresh_e:%(edge_type)s {branch: branch, branch_level: branch_level, status: status, from: from}]->(b)
+SET fresh_e.to = to
+WITH a, branch, status, from, to, b, %(id_func)s(fresh_e) AS e_id_to_keep
+CALL {
+    WITH a, branch, status, from, to, b, e_id_to_keep
+    MATCH (a)-[e:%(edge_type)s]->(b)
+    WHERE %(id_func)s(e) <> e_id_to_keep
+    AND e.branch = branch AND e.status = status AND e.from = from
+    AND (e.to = to OR (e.to IS NULL AND to IS NULL))
+    DELETE e
+}
+        """ % {"edge_type": self.edge_type.value, "id_func": db.get_id_function_name()}
+        self.add_to_query(query)
+
+
+class DeleteDuplicateIsVisibleEdgesQuery(DeleteDuplicateBooleanEdgesQuery):
+    name = "delete_duplicate_is_visible_edges"
+    type = QueryType.WRITE
+    insert_return = False
+    edge_type = DatabaseEdgeType.IS_VISIBLE
+
+
+class DeleteDuplicateIsProtectedEdgesQuery(DeleteDuplicateBooleanEdgesQuery):
+    name = "delete_duplicate_is_protected_edges"
+    type = QueryType.WRITE
+    insert_return = False
+    edge_type = DatabaseEdgeType.IS_PROTECTED
+
+
+class Migration020(GraphMigration):
+    """
+    1. Find duplicate edges. These can be duplicated if multiple AttributeValue nodes with the same value exist b/c of concurrent
+        database updates.
+        a. (a:Attribute)-[e:HAS_VALUE]->(av:AttributeValue)
+            grouped by (a, e.branch, e.from, e.to, e.status, av.value, av.is_default) to determine the number of duplicates.
+        b. (a:Attribute)-[e:HAS_VALUE]->(b:Boolean)
+            grouped by (a, e.branch, e.from, e.status, b) to determine the number of duplicates.
+    2. For a given set of duplicate edges
+        a. delete all of the duplicate edges
+        b. merge one edge with the properties of the deleted edges
+    3. If there are any orphaned AttributeValue nodes after these changes, then delete them
+
+    This migration does not account for consolidating duplicated AttributeValue nodes because more might be created
+    in the future due to concurrent database updates. A migration to consolidate duplicated AttributeValue nodes
+    should be run when we find a way to stop duplicate AttributeValue nodes from being created
+    """
+
+    name: str = "020_delete_duplicate_edges"
+    minimum_version: int = 19
+    queries: Sequence[type[Query]] = [
+        DeleteDuplicateHasValueEdgesQuery,
+        DeleteDuplicateIsVisibleEdgesQuery,
+        DeleteDuplicateIsProtectedEdgesQuery,
+    ]
+
+    async def execute(self, db: InfrahubDatabase) -> MigrationResult:
+        # skip the transaction b/c it will run out of memory on a large database
+        return await self.do_execute(db=db)
+
+    async def validate_migration(self, db: InfrahubDatabase) -> MigrationResult:  # noqa: ARG002
+        result = MigrationResult()
+        return result

backend/infrahub/core/migrations/graph/m020_add_generate_template_attr.py renamed to backend/infrahub/core/migrations/graph/m021_add_generate_template_attr.py

@@ -15,8 +15,8 @@
     from infrahub.database import InfrahubDatabase
 
 
-class Migration020(InternalSchemaMigration):
-    name: str = "020_add_generate_template_attr"
+class Migration021(InternalSchemaMigration):
+    name: str = "021_add_generate_template_attr"
     minimum_version: int = 19
 
     @classmethod

backend/infrahub/core/migrations/shared.py

@@ -120,15 +120,17 @@ async def validate_migration(self, db: InfrahubDatabase) -> MigrationResult:
 
     async def execute(self, db: InfrahubDatabase) -> MigrationResult:
         async with db.start_transaction() as ts:
-            result = MigrationResult()
+            return await self.do_execute(db=ts)
 
-            for migration_query in self.queries:
-                try:
-                    query = await migration_query.init(db=ts)
-                    await query.execute(db=ts)
-                except Exception as exc:
-                    result.errors.append(str(exc))
-                    return result
+    async def do_execute(self, db: InfrahubDatabase) -> MigrationResult:
+        result = MigrationResult()
+        for migration_query in self.queries:
+            try:
+                query = await migration_query.init(db=db)
+                await query.execute(db=db)
+            except Exception as exc:
+                result.errors.append(str(exc))
+                return result
 
         return result
 

backend/infrahub/core/node/constraints/grouped_uniqueness.py

@@ -177,12 +177,25 @@ async def _check_one_schema(
         await self._check_results(updated_node=node, path_groups=path_groups, query_results=query.get_results())
 
     async def check(self, node: Node, at: Optional[Timestamp] = None, filters: Optional[list[str]] = None) -> None:
+        def _frozen_constraints(schema: MainSchemaTypes) -> frozenset[frozenset[str]]:
+            if not schema.uniqueness_constraints:
+                return frozenset()
+            return frozenset(frozenset(uc) for uc in schema.uniqueness_constraints)
+
         node_schema = node.get_schema()
-        schemas_to_check: list[MainSchemaTypes] = [node_schema]
+        include_node_schema = True
+        frozen_node_constraints = _frozen_constraints(node_schema)
+        schemas_to_check: list[MainSchemaTypes] = []
         if node_schema.inherit_from:
             for parent_schema_name in node_schema.inherit_from:
                 parent_schema = self.schema_branch.get(name=parent_schema_name, duplicate=False)
-                if parent_schema.uniqueness_constraints:
-                    schemas_to_check.append(parent_schema)
+                if not parent_schema.uniqueness_constraints:
+                    continue
+                schemas_to_check.append(parent_schema)
+                frozen_parent_constraints = _frozen_constraints(parent_schema)
+                if frozen_node_constraints <= frozen_parent_constraints:
+                    include_node_schema = False
+        if include_node_schema:
+            schemas_to_check.append(node_schema)
         for schema in schemas_to_check:
             await self._check_one_schema(node=node, node_schema=schema, at=at, filters=filters)

backend/infrahub/core/query/attribute.py

@@ -66,6 +66,8 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> No
         query = """
         MATCH (a:Attribute { uuid: $attr_uuid })
         MERGE (av:%(labels)s { %(props)s } )
+        WITH av, a
+        LIMIT 1
         CREATE (a)-[r:%(rel_label)s { branch: $branch, branch_level: $branch_level, status: "active", from: $at }]->(av)
         """ % {"rel_label": self.attr._rel_to_value_label, "labels": ":".join(labels), "props": ", ".join(prop_list)}