Source Code Documentation

This page documents the source code of the graph-native LPG normalization.

The GitHub repository is structured as follows:

• docs/: The documentation of the source code of this repository. Basis for the documentation.
• gnfd/: A Python package that implements the Graph Object Functional Dependencies
• graphs/: Contains graphs that are normalized as part of the evaluation
• out/: Contains the output of the evaluation as CSV files
• tests/: Contains Python pytests for the dependencies and the normalization

Additionally, three Jupyter Notebooks are provided:

• structural_metrics_tables_and_figures.ipynb is provided for the further analysis of the evaluation results of the structural metrics.
• query_experiment.ipynb is provided for the analysis of the query experiment.
• normalization-skavantzos-link.ipynb computes same metrics as our evaluation scenarios normalized using the method described in ¹

In the following, the Python source code is documented.

Python Documentation

normalize.py

normalize

perform_graph_native_normalization(driver: Driver, database, provided_dependencies: DependencySet, dep_filter: str = 'all') -> (DependencySet, list[str])

Performs graph-native normalization under consideration of the provided parameters.

Parameters:

• driver (Driver) –

  The connection to the graph database

• database –

  The name of the database in which the to be normalized graph is contained. Supported databases are: "neo4j" and "memgraph"

• provided_dependencies (DependencySet) –

  The dependencies to be considered for the normalization

• dep_filter (str, default: 'all' ) –

  Whether only a subset of dependencies should be used. Possible values: "node-left", "edge-left", "within-node", "within-go", "between-go", "all". Defaults to "all".

Returns:

• (DependencySet, list[str]) –

Source code in normalize.py

def perform_graph_native_normalization(
    driver: Driver,
    database,
    provided_dependencies: DependencySet,
    dep_filter: str = "all",
) -> (DependencySet, list[str]):
    """
    Performs graph-native normalization under consideration of the provided parameters.

    :param driver: The connection to the graph database
    :type driver: neo4j.Driver
    :param database: The name of the database in which the to be normalized graph is contained. Supported databases are: ``"neo4j"`` and ``"memgraph"``
    :type: str
    :param provided_dependencies: The dependencies to be considered for the normalization
    :type: gnfd.DependencySet
    :param dep_filter: Whether only a subset of dependencies should be used. Possible values: ``"node-left"``, ``"edge-left"``, ``"within-node"``, ``"within-go"``, ``"between-go"``, ``"all"``. Defaults to ``"all"``.
    :type: str
    :return:
    """

    """A local copy of the provided dependencies that, e.g., may be filtered."""
    deps = provided_dependencies

    """A list of strings of queries that create indices"""
    index_queries: set[str] = set()
    """A list of strings of the queries that perform the transformations"""
    transformation_queries: set[str] = set()
    """A list of strings of the queries that remove properties"""
    remove_queries: set[str] = set()
    """A list of strings of the queries that delete graph objects"""
    delete_queries: set[str] = set()

    """The set of dependencies after all transformations have been applied."""
    transformed_deps: DependencySet
    """A list of the string representations of the transformed dependencies."""
    transformed_deps_list: list[str] = []

    applied_transformations: list[str] = []

    def _apply_transformation_query(query: str):
        """Runs a query string on the graph connected through :any:`driver`.

        :param query: The to be run query.
        :type query: str"""
        with driver.session(database=database) as session:
            responsive = False
            while not responsive:
                responsive = True
                try:
                    session.run(query)
                except neo4j.exceptions.TransientError:
                    responsive = False
                    time.sleep(1)


    def validate_dep(dep):
        """Validates whether a functional dependency holds"""
        if dep.is_trivial:
            return # The dependency is technically valid, although not minimal!

        with driver.session(database=database) as session:
            query = f"""
{dep.pattern.to_gql_match_where_string()}
WITH DISTINCT
{",".join(map(lambda ref: str(ref)+" AS "+pascalcase(str(ref)), dep.left.union(dep.right)))}
WITH
{",".join(map(lambda ref: pascalcase(str(ref)), dep.left))},
COUNT([{",".join(map(lambda ref: pascalcase(str(ref)), dep.right))}]) AS card
RETURN avg(card) AS res

"""
            res = session.run(query)
            record = res.single()
            if record is not None:
                if record["res"] != 1:
                    raise ValueError(f'The dependency "{str(dep)}" is not functional!')

    # Phase 0: Filter deps according to parameter from evaluation.
    logging.info("Filter dependencies")
    match dep_filter:
        case "within-node":
            deps = DependencySet(filter(lambda dep: dep.is_within_node, deps))
        case "within-go":
            deps = DependencySet(filter(lambda dep: dep.is_within_graph_object, deps))
        case "between-go":
            deps = DependencySet(filter(lambda dep: dep.is_inter_graph_object, deps))
        case "node-left":
            deps = DependencySet(
                filter(
                    lambda dep: dep.is_within_node
                    or isinstance(next(iter(dep.left)).get_graph_object(), Node),
                    deps,
                )
            )
        case "edge-left":
            deps = DependencySet(
                filter(
                    lambda dep: isinstance(
                        next(iter(dep.left)).get_graph_object(), Edge
                    ),
                    deps,
                )
            )

    if len(deps) == 0:
        return (
            provided_dependencies,
            applied_transformations,
        )  # Nothing will happen --> Return original dependencies

    i = 0

    sorted_deps = list(deps)
    sorted_deps.sort()

    for dep in sorted_deps:
        validate_dep(dep)

        if dep.is_inter_graph_object:
            inter_dep = dep
            left_gos: set[GraphObject] = set(
                map(lambda ref: ref.get_graph_object(), dep.left)
            )

            if len(left_gos) == 1:  # Multiple GOs are not supported for the left side
                left_go = left_gos.pop()
                if isinstance(left_go, Edge):
                    edge = left_go
                    for right_ref in dep.right:
                        left_references: set[Reference] = set(
                            filter(lambda ref: ref.is_property_variable, dep.left)
                        )

                        if (
                            right_ref.is_property_variable
                            and len(left_references) > 0
                            and isinstance(right_ref.get_graph_object(), Node)
                            and (
                                right_ref.get_graph_object() is left_go.src
                                or right_ref.get_graph_object() is left_go.tgt
                            )
                        ):
                            logging.info("between-ep-np")
                            assert isinstance(right_ref.get_graph_object(), Node)

                            merge_key_elements = list(
                                map(str, left_references.union({right_ref}))
                            )
                            merge_key_elements.sort()
                            within_merge_key: str = ",".join(merge_key_elements)
                            new_props = list(left_references.union({right_ref}))
                            new_props.sort(key=str)
                            new_label: str = pascalcase(within_merge_key)

                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE CONSTRAINT IF NOT EXISTS FOR (newNode:{new_label}) REQUIRE (newNode.{", newNode.".join(map(pascalcase, map(str, left_references)))}) IS UNIQUE"
                                )
                            else:
                                assert database == "memgraph"
                                index_queries.add(f"""CREATE INDEX ON :{new_label}({",".join(map(pascalcase, map(str, left_references)))})""")

                            orig_edge_label = next(iter(edge.labels))
                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE INDEX IF NOT EXISTS FOR (n:{orig_edge_label}) ON (n.{", n.".join(edge.properties)})"
                                )
                                index_queries.add(
                                    f"CREATE INDEX IF NOT EXISTS FOR ()-[e:{orig_edge_label}]-() ON (e.{", e.".join(edge.properties)})"
                                )
                                if len(edge.src.properties) > 0:
                                    index_queries.add(
                                        f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(edge.src.labels)}) ON (n.{", n.".join(edge.src.properties)})"
                                    )
                                if len(edge.tgt.properties) > 0:
                                    index_queries.add(
                                        f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(edge.tgt.labels)}) ON (n.{", n.".join(edge.tgt.properties)})"
                                    )
                            else:
                                assert database == "memgraph"
                                index_queries.add(
                                    f"CREATE INDEX on :{orig_edge_label}({",".join(edge.properties)})"
                                )
                                index_queries.add(
                                    f"CREATE EDGE INDEX ON :{orig_edge_label} ({next(iter(edge.properties))})"
                                )


                            new_properties: str = ", ".join(
                                map(
                                    lambda ref: f"{pascalcase(ref)} : {ref}",
                                    map(str, new_props),
                                )
                            )

                            reify_and_extract_to_new_node(
                                edge,
                                i,
                                dep,
                                new_label,
                                new_properties,
                                new_props,
                                transformation_queries,
                            )
                            rand = str(uuid.uuid4())[:8]
                            node = right_ref.get_graph_object()

                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(node.labels)}) ON (n.`{rand}`)"
                                )
                            else:
                                index_queries.add(
                                    f"CREATE INDEX ON :{next(iter(node.labels))}(`{rand}`)"
                                )

                            index_queries.add(
f"""
{dep.pattern.to_gql_match_where_string()} 
WITH DISTINCT {node.symbol}
SET {node.symbol}.`{rand}`="{rand}"
"""
                            )
                            # Remove old redundant properties in the end
                            cleanup_pattern = (
                                inter_dep.pattern.to_gql_match_where_string().split(
                                    "WHERE"
                                )[0]
                            )
                            remove_queries.add(
    f"""
{cleanup_pattern} 
WHERE {node.symbol}.`{rand}`="{rand}"
WITH DISTINCT {node.symbol}
REMOVE {right_ref}
REMOVE {node.symbol}.`{rand}`
"""
                            )


                            delete_queries.add(
                                f"""
MATCH ({"".join(map(lambda lab: f":{lab}", edge.src.labels))})-[{edge.symbol}{"".join(map(lambda lab: f":{lab}", edge.labels))}]->({"".join(map(lambda lab: f":{lab}", edge.tgt.labels))})
WITH DISTINCT {edge.symbol}
DELETE {edge.symbol}"""
                            )


                            remove_queries.add(
                                f"""
MATCH ({edge.symbol}) - [:{new_label.upper()}]->(x{i})
REMOVE {", ".join(map(str, left_references))}"""
                            )  # Connect normalized nodes with reified nodes and remove redundant properties

                            transformed_deps_list.append(
                                f"""
(x{i}:{new_label}:{"&".join(map(pascalcase, map(str, left_references.union({right_ref}))))})
::
{",".join(map(lambda ref: f"x{i}.{pascalcase(ref)}", map(str, left_references)))}
=>x{i}""".replace(
                                    " ", ""
                                ).replace(
                                    "\n", ""
                                )
                            )

                            applied_transformations.append("between-ep-np")

                        elif (
                            right_ref.is_graph_object_variable
                            and len(left_references) > 0
                            and isinstance(right_ref.get_graph_object(), Node)
                            and (
                                right_ref.get_graph_object() is left_go.src
                                or right_ref.get_graph_object() is left_go.tgt
                            )
                        ):
                            logging.info("between-ep-n")

                            merge_key_elements = list(map(str, left_references))
                            merge_key_elements.sort()
                            within_merge_key: str = ",".join(merge_key_elements)
                            new_props = list(left_references)
                            new_props.sort(key=str)
                            new_label: str = pascalcase(within_merge_key)

                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE CONSTRAINT IF NOT EXISTS FOR (newNode:{new_label}) REQUIRE (newNode.{", newNode.".join(map(pascalcase, map(str, left_references)))}) IS UNIQUE"
                                )
                            else:
                                assert database == "memgraph"
                                index_queries.add(f"""CREATE INDEX ON :{new_label}({",".join(map(pascalcase, map(str, left_references)))})""")

                            orig_edge_label = next(iter(edge.labels))
                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE INDEX IF NOT EXISTS FOR (n:{orig_edge_label}) ON (n.{", n.".join(edge.properties)})"
                                )
                                index_queries.add(
                                    f"CREATE INDEX IF NOT EXISTS FOR ()-[e:{orig_edge_label}]-() ON (e.{", e.".join(edge.properties)})"
                                )
                                if len(edge.src.properties) > 0:
                                    index_queries.add(
                                        f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(edge.src.labels)}) ON (n.{", n.".join(edge.src.properties)})"
                                    )
                                if len(edge.tgt.properties) > 0:
                                    index_queries.add(
                                        f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(edge.tgt.labels)}) ON (n.{", n.".join(edge.tgt.properties)})"
                                    )
                            else:
                                assert database == "memgraph"
                                index_queries.add(
                                    f"CREATE INDEX on :{orig_edge_label}({",".join(edge.properties)})"
                                )
                                index_queries.add(
                                    f"CREATE EDGE INDEX ON :{orig_edge_label} ({next(iter(edge.properties))})"
                                )


                            new_properties: str = ", ".join(
                                map(
                                    lambda ref: f"{pascalcase(ref)} : {ref}",
                                    map(str, new_props),
                                )
                            )

                            reify_and_extract_to_new_node(
                                edge,
                                i,
                                dep,
                                new_label,
                                new_properties,
                                new_props,
                                transformation_queries,
                            )

                            delete_queries.add(
                                f"""
MATCH ({"".join(map(lambda lab: f":{lab}", edge.src.labels))})-[{edge.symbol}{"".join(map(lambda lab: f":{lab}", edge.labels))}]->({"".join(map(lambda lab: f":{lab}", edge.tgt.labels))})
WITH DISTINCT {edge.symbol}
DELETE {edge.symbol}"""
                            )

                            remove_queries.add(
                                f"""
MATCH ({edge.symbol})-[:{new_label.upper()}]->(x{i})
WITH DISTINCT {edge.symbol}
REMOVE {", ".join(map(str, left_references))}"""
                            )  # Connect normalized nodes with reified nodes and remove redundant properties

                            transformed_deps_list.append(
                                f"""
(x{i}:{new_label}:{"&".join(map(pascalcase, map(str, left_references.union({right_ref}))))})
::
{",".join(map(lambda ref: f"x{i}.{pascalcase(ref)}", map(str, left_references)))}
=>x{i}""".replace(
                                    " ", ""
                                ).replace(
                                    "\n", ""
                                )
                            )
                            applied_transformations.append("between-ep-n")

                elif isinstance(left_go, Node):
                    node = left_go
                    for right_ref in inter_dep.right:
                        left_is_go = (
                            len(
                                set(
                                    filter(
                                        lambda ref: "." not in str(ref), inter_dep.left
                                    )
                                )
                            )
                            > 0
                        )
                        left_references: set[Reference] = set(
                            filter(lambda ref: "." in str(ref), inter_dep.left)
                        )

                        if (
                            "." in str(right_ref)
                            and left_is_go
                            and isinstance(right_ref.get_graph_object(), Edge)
                            and (
                                right_ref.get_graph_object().src is node
                                or right_ref.get_graph_object().tgt is node
                            )
                        ):
                            logging.info("between-n-ep")

                            edge = right_ref.get_graph_object()
                            assert isinstance(edge, Edge)

                            transformation_queries.add(
                                f"""
{inter_dep.pattern.to_gql_match_where_string()}
WITH DISTINCT {left_go.symbol}, {right_ref} AS {camelcase(str(right_ref))}
SET {left_go.symbol}.{pascalcase(str(right_ref))} = {camelcase(str(right_ref))}"""
                            )
                            remove_queries.add(
                                f"""
{inter_dep.pattern.to_gql_match_where_string()}
WITH DISTINCT {edge.symbol}
REMOVE {right_ref}"""
                            )
                            node.properties.add(pascalcase(str(right_ref)))
                            right_ref.get_graph_object().properties.remove(
                                str(right_ref).split(".")[1]
                            )
                            transformed_deps_list.append(
                                f"({node.symbol}:{"&".join(node.labels)}:{pascalcase(str(right_ref))})::{node.symbol}=>{node.symbol}.{pascalcase(str(right_ref))}"
                            )
                            applied_transformations.append("between-n-ep")

                        elif (
                            right_ref.is_property_variable
                            and not left_is_go
                            and isinstance(right_ref.get_graph_object(), Edge)
                            and (
                                right_ref.get_graph_object().src is node
                                or right_ref.get_graph_object().tgt is node
                            )
                        ):
                            logging.info("between-np-ep")

                            merge_key_elements = list(
                                map(str, left_references.union({right_ref}))
                            )
                            merge_key_elements.sort()
                            within_merge_key: str = ",".join(merge_key_elements)
                            new_props = list(left_references.union({right_ref}))
                            new_props.sort(key=str)
                            new_label: str = pascalcase(within_merge_key)

                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE CONSTRAINT IF NOT EXISTS FOR (newNode:{new_label}) REQUIRE (newNode.{", newNode.".join(map(pascalcase, map(str, left_references)))}) IS UNIQUE"
                                )
                            else:
                                assert database == "memgraph"
                                index_queries.add(
                                    f"""CREATE INDEX ON :{new_label}({",".join(map(pascalcase, map(str, left_references)))})""")


                            new_properties: str = ", ".join(
                                map(
                                    lambda ref: f"{pascalcase(ref)} : {ref}",
                                    map(str, new_props),
                                )
                            )

                            rand = str(uuid.uuid4())[:8]
                            if database == "neo4j":
                                index_queries.add(
                                    f"CREATE INDEX IF NOT EXISTS FOR (n:{next(iter(node.labels))}) ON (n.`{rand}`)"
                                )
                            else:
                                index_queries.add(
                                    f"CREATE INDEX ON :{next(iter(node.labels))}(`{rand}`)"
                                )


                            transformation_queries.add(
                                f"""
                            {dep.pattern.to_gql_match_where_string()} 
MERGE (newNode:{new_label} {{{new_properties}}})
MERGE ({node.symbol})-[:{new_label.upper()}]->(newNode)
WITH DISTINCT {node.symbol}
SET {node.symbol}.`{rand}`="{rand}"
"""
                            )

                            # Remove old redundant properties in the end
                            cleanup_pattern = (
                                inter_dep.pattern.to_gql_match_where_string().split(
                                    "WHERE"
                                )[0]
                            )
                            remove_queries.add(
                                f"""
{cleanup_pattern} 
WHERE {node.symbol}.`{rand}`="{rand}"
REMOVE {", ".join(map(str, left_references.union({right_ref})))}
WITH DISTINCT {node.symbol}
REMOVE {node.symbol}.`{rand}`
"""
                            )

                            right_ref.get_graph_object().properties -= {right_ref}
                            node.properties -= left_references

                            transformed_deps_list.append(
                                f"""
(x{i}:{new_label}:{"&".join(map(pascalcase, map(str, left_references.union({right_ref}))))})
::
{",".join(map(lambda ref: f"x{i}.{pascalcase(ref)}", map(str, left_references)))}
=>x{i}""".replace(
                                    " ", ""
                                ).replace(
                                    "\n", ""
                                )
                            )

                            applied_transformations.append("between-np-ep")

        elif dep.is_within_graph_object:
            # First filter References that are Graph Object IDs. We don't need them here as their occurrence is a sign for structurally implied or to limiting dep.s.
            left_references: set[Reference] = set(
                filter(lambda ref: ref.is_property_variable, dep.left)
            )
            right_references: set[Reference] = set(
                filter(lambda ref: ref.is_property_variable, dep.right)
            )
            all_references: set[Reference] = left_references.union(right_references)

            merge_key_elements = list(map(str, all_references))
            merge_key_elements.sort()
            within_merge_key: str = ",".join(merge_key_elements)
            new_props = list(right_references.union(left_references))
            new_props.sort(key=str)
            new_label: str = pascalcase(within_merge_key)

            i += 1

            # # # # # # # # # #
            #  ψ_L1 (psi_L1)  #
            # # # # # # # # # #
            if (
                dep.is_within_node
                and len(left_references) > 0
                and len(right_references) > 0
            ):
                logging.info("within-n")
                node: Node = dep.right.pop().get_graph_object()

                if database == "neo4j":
                    index_queries.add(
                        f"CREATE CONSTRAINT IF NOT EXISTS FOR (newNode:{new_label}) REQUIRE (newNode.{", newNode.".join(map(pascalcase, map(str, left_references)))}) IS UNIQUE"
                    )
                else:
                    assert database == "memgraph"
                    index_queries.add(
                        f"""CREATE INDEX ON :{new_label}({",".join(map(pascalcase, map(str, left_references)))})""")

                new_properties: str = ", ".join(
                    map(lambda ref: f"{pascalcase(ref)} : {ref}", map(str, new_props))
                )

                rand = str(uuid.uuid4())[:8]
                if database == "neo4j":
                    index_queries.add(
                        f"CREATE INDEX IF NOT EXISTS FOR (n:{next(iter(node.labels))}) ON (n.`{rand}`)"
                    )
                else:
                    index_queries.add(
                        f"CREATE INDEX ON :{next(iter(node.labels))}(`{rand}`)"
                    )

                transformation_queries.add(
                    f"""
{dep.pattern.to_gql_match_where_string()} 
MERGE (newNode:{new_label} {{{new_properties}}})
MERGE ({node.symbol})-[:{new_label.upper()}]->(newNode)
WITH DISTINCT {node.symbol}
SET {node.symbol}.`{rand}`="{rand}"
"""
                )

                # Remove old redundant properties in the end
                cleanup_pattern = dep.pattern.to_gql_match_where_string().split(
                    "WHERE"
                )[0]
                remove_queries.add(
                    f"""
{cleanup_pattern} 
WITH DISTINCT {node.symbol}
WHERE {node.symbol}.`{rand}`="{rand}"
REMOVE {node.symbol}.`{rand}`
REMOVE {", ".join(map(str, right_references.union(left_references)))}
"""
                )

                dep.pattern.properties -= all_references

                transformed_deps_list.append(
                    f"""
(x{i}:{new_label}:{"&".join(map(pascalcase, map(str, right_references.union(left_references))))})
::
{",".join(map(lambda ref: f"x{i}.{pascalcase(ref)}", map(str, left_references)))}
=>x{i}""".replace(
                        " ", ""
                    ).replace(
                        "\n", ""
                    )
                )

                applied_transformations.append("within-n")

            # # # # # # # # # #
            #  ψ_L2 (psi_L2)  #
            # # # # # # # # # #
            elif (
                dep.is_within_edge
                and len(left_references) > 0
                and len(right_references) > 0
            ):  # ψ_L2 (psi_L2)  --> Reification
                logging.info("Within e")

                edge: Edge = dep.right.pop().get_graph_object()

                if database == "neo4j":
                    index_queries.add(
                        f"CREATE CONSTRAINT IF NOT EXISTS FOR (newNode:{new_label}) REQUIRE (newNode.{", newNode.".join(map(pascalcase, map(str, left_references)))}) IS UNIQUE"
                    )
                else:
                    assert database == "memgraph"
                    index_queries.add(
                        f"""CREATE INDEX ON :{new_label}({",".join(map(pascalcase, map(str, left_references)))})""")

                orig_edge_label = next(iter(edge.labels))
                if database == "neo4j":
                    index_queries.add(
                        f"CREATE INDEX IF NOT EXISTS FOR (n:{orig_edge_label}) ON (n.{", n.".join(edge.properties)})"
                    )
                    index_queries.add(
                        f"CREATE INDEX IF NOT EXISTS FOR ()-[e:{orig_edge_label}]-() ON (e.{", e.".join(edge.properties)})"
                    )
                    if len(edge.src.properties) > 0:
                        index_queries.add(
                            f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(edge.src.labels)}) ON (n.{", n.".join(edge.src.properties)})"
                        )
                    if len(edge.tgt.properties) > 0:
                        index_queries.add(
                            f"CREATE INDEX IF NOT EXISTS FOR (n:{":".join(edge.tgt.labels)}) ON (n.{", n.".join(edge.tgt.properties)})"
                        )
                else:
                    assert database == "memgraph"
                    index_queries.add(
                        f"CREATE INDEX on :{orig_edge_label}({",".join(edge.properties)})"
                    )
                    index_queries.add(
                        f"CREATE EDGE INDEX ON :{orig_edge_label}({next(iter(edge.properties))})"
                    )


                new_properties: str = ", ".join(
                    map(lambda ref: f"{pascalcase(ref)} : {ref}", map(str, new_props))
                )

                # Reification + create new node
                reify_and_extract_to_new_node(
                    edge,
                    i,
                    dep,
                    new_label,
                    new_properties,
                    new_props,
                    transformation_queries,
                )

                delete_queries.add(
                    f"""
MATCH ({"".join(map(lambda lab: f":{lab}", edge.src.labels))})-[{edge.symbol}{"".join(map(lambda lab: f":{lab}", edge.labels))}]->({"".join(map(lambda lab: f":{lab}", edge.tgt.labels))})
REMOVE {", ".join(map(str, all_references))}
WITH DISTINCT {edge.symbol}
DELETE {edge.symbol}"""
                )

                remove_queries.add(
                    f"""
MATCH ({edge.symbol})-[:{new_label.upper()}]->(x{i})
WITH DISTINCT {edge.symbol}
REMOVE {", ".join(map(str, all_references))}"""
                )  # Connect normalized nodes with reified nodes and remove redundant properties

                transformed_deps_list.append(
                    f"""
(x{i}:{new_label}:{"&".join(map(pascalcase, map(str, right_references.union(left_references))))})
::
{",".join(map(lambda ref: f"x{i}.{pascalcase(ref)}", map(str, left_references)))}
=>x{i}""".replace(
                        " ", ""
                    ).replace(
                        "\n", ""
                    )
                )

                applied_transformations.append("within-e")

        i += 1

    for query in tqdm(index_queries, desc="  Indices"):
        logging.info(query)
        _apply_transformation_query(query)
    for query in tqdm(transformation_queries, desc="  Query"):
        _apply_transformation_query(query)
    for query in tqdm(remove_queries, desc="  Cleanup (REMOVE)"):
        _apply_transformation_query(query)
    for query in tqdm(delete_queries, desc="  Cleanup (DELETE)"):
        _apply_transformation_query(query)

    transformed_deps = DependencySet.from_string_list(transformed_deps_list)

    return transformed_deps, applied_transformations

GNFD Package

gnfd

DependencySet

Bases: set[GOFD]

Source code in gnfd/__init__.py

class DependencySet(set[GOFD]):
    @classmethod
    def from_string_list(cls, lst: list[str]) -> DependencySet:
        """Creates a :class:`DependencySet` from a list of dependencies encoded as strings."""
        return cls(map(GOFD.from_string, lst))


    def is_in_global_normal_form(self) -> bool:
        """:returns: :any:`True` when there is no inter-graph dependency, :any:`False` otherwise."""
        return sum(map(lambda dep: dep.is_inter_graph_object, self)) == 0

    @property
    def lp_suitable(self) -> bool:
        """:returns: :any:`True` when LP normalization (cf. <https://doi.org/10.1007/s00778-025-00902-2>) could be performed on this graph as its dependencies only target nodes and there are no inter graph dependencies."""
        return self.is_in_global_normal_form() and reduce(operator.and_,
                                                          map(lambda dep: dep.is_within_node, iter(self)))

lp_suitable: bool property

Returns:

• bool –

  :any:True when LP normalization (cf. https://doi.org/10.1007/s00778-025-00902-2) could be performed on this graph as its dependencies only target nodes and there are no inter graph dependencies.

from_string_list(lst: list[str]) -> DependencySet classmethod

Creates a :class:DependencySet from a list of dependencies encoded as strings.

Source code in gnfd/__init__.py

@classmethod
def from_string_list(cls, lst: list[str]) -> DependencySet:
    """Creates a :class:`DependencySet` from a list of dependencies encoded as strings."""
    return cls(map(GOFD.from_string, lst))

is_in_global_normal_form() -> bool

Returns:

• bool –

  :any:True when there is no inter-graph dependency, :any:False otherwise.

Source code in gnfd/__init__.py

def is_in_global_normal_form(self) -> bool:
    """:returns: :any:`True` when there is no inter-graph dependency, :any:`False` otherwise."""
    return sum(map(lambda dep: dep.is_inter_graph_object, self)) == 0

Edge

Bases: GraphObject, Pattern, ABC

Represents an abstract LPG edge without a direction.

Parameters:

• symbol (str, default: '' ) –

  The symbol of the variable of the edge.

• labels (set[str] | None, default: None ) –

  The labels of the edge.

• properties (set[str] | None, default: None ) –

  The properties of the edge.

• src_pattern (Pattern | None, default: None ) –

  The source :any:Node of the edge.

• tgt_pattern (Pattern | None, default: None ) –

  The target :any:Node of the edge.

Source code in gnfd/gnfd.py

class Edge(GraphObject, Pattern, abc.ABC):
    """Represents an abstract LPG edge without a direction.

    :param symbol: The symbol of the variable of the edge.
    :param labels: The labels of the edge.
    :param properties: The properties of the edge.
    :param src_pattern: The source :any:`Node` of the edge.
    :param tgt_pattern: The target :any:`Node` of the edge.
    """

    def __init__(
        self,
        symbol: str = "",
        labels: set[str] | None = None,
        properties: set[str] | None = None,
        src_pattern: Pattern | None = None,
        tgt_pattern: Pattern | None = None,
    ):
        self.src_pattern: Pattern | None = src_pattern
        self.tgt_pattern: Pattern | None = tgt_pattern
        super().__init__(symbol, labels, properties)

    def contains_var(self, symbol: str) -> bool:
        return (
            symbol == self.symbol
            or self.src_pattern.contains_var(symbol)
            or self.tgt_pattern.contains_var(symbol)
        )

    def contains_property(self, symbol: str, key: str):
        return (
            (self.contains_var(symbol) and key in self.properties)
            or self.src_pattern.contains_property(symbol, key)
            or self.tgt_pattern.contains_property(symbol, key)
        )

    def to_clingo_tableau(self, lhs, constants: bool = True) -> str:
        res = ""
        edge_symbol = self.clingo_symbol()

        def get_symbol(prop_symbol, suffix="") -> str:
            if f"{self.symbol}.{prop_symbol}" in map(str, lhs):
                return f"{self.symbol}_{prop_symbol}"
            else:
                return f"{self.symbol}_{prop_symbol}_{suffix}"

        for i in ["1", "2"]:
            if self not in map(lambda ref: ref.reference, lhs):
                edge_symbol += i  # the symbol of the node occurs not on the left side of a dep, so it must be different
            if constants:
                res += f"enc({self.src.clingo_symbol(lhs,i).lower()},edge__,{self.tgt.clingo_symbol(lhs,i).lower()},{self.clingo_symbol(lhs,i).lower()}).\n"
                res += "".join(
                    map(
                        lambda
                            lab: f"enc({self.clingo_symbol(lhs,i).lower()},lab__,str__{lab}, {uuid.uuid4().hex[:7].lower()}).\n",
                        self.labels,
                    )
                )
                res += "".join(
                    map(
                        lambda
                            prop: f"enc({self.clingo_symbol(lhs,i).lower()},{prop}__,str__{prop}, {uuid.uuid4().hex[:7].lower()}).\n",
                        self.properties,
                    )
                )
            else:
                res += f"enc({self.src.clingo_symbol(lhs,i).upper()},Edge__,{self.tgt.clingo_symbol(lhs,i).upper()},{self.clingo_symbol(lhs,i).upper()}).\n"
                res += "".join(
                    map(
                        lambda
                            lab: f"enc({self.clingo_symbol(lhs,i).upper()},lab__,Str__{lab}, {uuid.uuid4().hex[:7].upper()}).\n",
                        self.labels,
                    )
                )
                res += "".join(
                    map(
                        lambda
                            prop: f"enc({self.clingo_symbol(lhs,i).upper()},{prop}__,Str__{prop}, {uuid.uuid4().hex[:7].upper()}).\n",
                        self.properties,
                    )
                )
     #   res += self.src_pattern.to_clingo()
     #   res += self.tgt_pattern.to_clingo()
        return res

    def to_gql_match_where_tuple(self) -> tuple[str, str]:
        tup_src_pattern = self.src_pattern.to_gql_match_where_tuple()
        tup_tgt_pattern = self.tgt_pattern.to_gql_match_where_tuple()

        where = f"{" AND ".join(map(lambda key: f"{self.symbol}.{key} IS NOT NULL", self.properties))}"
        if len(where) > 0 and len(tup_src_pattern[1]) > 0:
            where += " AND "
        where += tup_src_pattern[1]
        if len(where) > 0 and len(tup_tgt_pattern[1]) > 0:
            where += " AND "
        where += tup_tgt_pattern[1]

        label_delim: str = ""
        if len(self.labels) > 0:
            label_delim = ":"

        middle = f"({self.src.symbol})-[{self.symbol}{label_delim}{":".join(self.labels)}]->({self.tgt.symbol})"

        # Don't match with every node! --> would result in a match with the cartesian product
        src = tup_src_pattern[0]
        tgt = tup_tgt_pattern[0]

        if src == "()" and tgt == "()":
            return middle, where
        elif src == "()":
            return f"{middle},{tgt}", where
        elif tgt == "()":
            return f"{src},{middle}", where

        return f"{src},{middle},{tgt}", where

    def get_graph_object_with_symbol(self, symbol: str) -> GraphObject | None:
        if self.symbol == symbol and symbol != "":
            return self
        else:
            sgo = self.src_pattern.get_graph_object_with_symbol(symbol)
            tgo = self.tgt_pattern.get_graph_object_with_symbol(symbol)
            if sgo is not None:
                return sgo
            else:
                return tgo

    def get_graph_objects(self) -> set[GraphObject]:
        res = set()
        res.add(self)
        res = res.union(self.src_pattern.get_graph_objects())
        res = (
            res.union(self.tgt_pattern.get_graph_objects()))
        return res


    def less_than(self, other):
        if not isinstance(other, Edge):
            return False
        return (((other.labels.issubset(self.labels) and other.properties.issubset(self.properties)) or
                (other.labels == self.labels and other.properties.issubset(self.properties)) or
        (other.labels.issubset(self.labels) and other.properties == self.properties)) and
            ((self.src.less_than(other.src) and self.tgt.less_than(other.tgt) or
            (self.src.equals(other.src) and self.tgt.less_than(other.tgt) or
            (self.src.less_than(other.src) and self.tgt.equals(other.tgt))
            ))))

    @property
    @abstractmethod
    def src(self) -> Node:
        """:returns the source :any:`Node` of the edge"""
        pass

    @property
    @abstractmethod
    def tgt(self) -> Node:
        """:returns the target :any:`Node` of the edge"""
        pass

    def go_equals(self, other: GraphObject) -> bool:
        return (
            isinstance(other, Edge)
            and self.src.go_equals(other.src)
            and self.tgt.go_equals(other.tgt)
            and self.labels == other.labels
            and self.properties == other.properties
        )

    def minimal_pattern_intersections(
        self, other: Pattern
    ) -> list[tuple[Pattern, set[tuple[str, str]]]]:
        res: list[tuple[Pattern, set[tuple[str, str]]]] = []

        if isinstance(other, Node): # case 2
            return other.minimal_pattern_intersections(self)

        elif isinstance(other, Edge): # case 4
            for mpi in self.minimal_pattern_intersections(other.src_pattern): # case 4.1
                clone = copy.deepcopy(self)
                clone.src_pattern = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (clone, rep)
                res.append(tup)
            for mpi in self.minimal_pattern_intersections(other.tgt_pattern): # case 4.2
                clone = copy.deepcopy(self)
                clone.tgt_pattern = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (clone, rep)
                res.append(tup)
            for mpi1 in self.minimal_pattern_intersections(other.src_pattern): # case 4.3
                for mpi2 in self.minimal_pattern_intersections(other.tgt_pattern):
                    clone = copy.deepcopy(self)
                    clone.src_pattern = mpi1[0]
                    clone.tgt_pattern = mpi2[0]
                    clone.labels = clone.labels.union(other.labels)
                    clone.properties = clone.properties.union(other.properties)
                    rep = set()
                    rep = rep.union(mpi1[1])
                    rep = rep.union(mpi2[1])
                    rep.add((self.symbol, other.symbol))
                    tup = (clone, rep)
                    res.append(tup)

        elif isinstance(other, PatternConcat):
            for mpi in self.minimal_pattern_intersections(other.left): # case 5.1
                other_clone = copy.deepcopy(other)
                other_clone.left = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)
            for mpi in self.minimal_pattern_intersections(other.right): # case 5.2
                other_clone = copy.deepcopy(other)
                other_clone.right = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)

        return res

src: Node abstractmethod property

Returns:

• Node –

  any:Node of the edge

tgt: Node abstractmethod property

Returns:

• Node –

  any:Node of the edge

GOFD

Denotes a GO-FD that consists of a :any:Pattern and sets of :any:Reference that denote the right and left side of the descriptor of the GO-FD. GNFDs are ordered by their pattern.

Source code in gnfd/gnfd.py

@total_ordering
class GOFD:
    """Denotes a GO-FD that consists of a :any:`Pattern` and sets of :any:`Reference` that denote the right and left side of the descriptor of the GO-FD.
    GNFDs are ordered by their pattern."""

    def __init__(
        self,
        pattern: Pattern,
        left: set[Reference],
        right: set[Reference],
    ):
        self.pattern: Pattern = pattern
        self.left: set[Reference] = left
        self.right: set[Reference] = right

    def __str__(self):
        return f"{str(self.pattern)}::{",".join(map(str, self.left))}=>{",".join(map(str, self.right))}"

    @staticmethod
    def from_string(string: str):
        """Parse a GO-FD from a string.

        :param string: The string representation of the GO-FD to create. Relies on the grammar shown in `gnfd.g4`.
        """

        # ANTLR has been run in this folder using the command `antlr4 -Dlanguage=Python3 -visitor gnfd.g4`.
        input_stream = InputStream(string)
        lexer = gnfdLexer(input_stream)
        stream = CommonTokenStream(lexer)
        parser = gnfdParser(stream)
        tree = parser.dependency()
        listener = _GNFDListener()

        walker = ParseTreeWalker()
        walker.walk(listener, tree)

        return listener.dependency

    @property
    def is_within_graph_object(self):
        return self.is_within_node or self.is_within_edge

    @property
    def is_within_node(self):
        gos = set(map(lambda ref: ref.get_graph_object(), self.left.union(self.right)))
        if len(gos) == 1:
            return isinstance(gos.pop(), Node)
        else:
            return False

    @property
    def is_within_edge(self):
        gos = set(map(lambda ref: ref.get_graph_object(), self.left.union(self.right)))
        if len(gos) == 1:
            return isinstance(gos.pop(), Edge)
        else:
            return False

    @property
    def is_inter_graph_object(self):
        return not self.is_within_graph_object

    def to_latex(self):
        return f"{str(self.pattern)}::{",".join(map(str, self.left))}\\determ {",".join(map(str, self.right))}"

    @property
    def is_trivial(self):
        return len(set(map(str, self.right)).intersection(set(map(str, self.left)))) > 0

    # Comparison is based in pattern!

    def __eq__(self, other):
        if not isinstance(other, GOFD):
            return NotImplemented
        return self.pattern.equals(other.pattern)

    def __lt__(self, other):
        if not isinstance(other, GOFD):
            return NotImplemented
        return self.pattern.less_than(other.pattern)

    def __hash__(self):
        return hash(str(self))

from_string(string: str) staticmethod

Parse a GO-FD from a string.

Parameters:

• string (str) –

  The string representation of the GO-FD to create. Relies on the grammar shown in gnfd.g4.

Source code in gnfd/gnfd.py

@staticmethod
def from_string(string: str):
    """Parse a GO-FD from a string.

    :param string: The string representation of the GO-FD to create. Relies on the grammar shown in `gnfd.g4`.
    """

    # ANTLR has been run in this folder using the command `antlr4 -Dlanguage=Python3 -visitor gnfd.g4`.
    input_stream = InputStream(string)
    lexer = gnfdLexer(input_stream)
    stream = CommonTokenStream(lexer)
    parser = gnfdParser(stream)
    tree = parser.dependency()
    listener = _GNFDListener()

    walker = ParseTreeWalker()
    walker.walk(listener, tree)

    return listener.dependency

GraphObject

Bases: ABC

Abstract class representing an LPG graph object, i.e., a :any:Edge or a :any:Node.

Parameters:

• symbol (str, default: '' ) –

  The symbol of the variable of the graph object.

• labels (set[str] | None, default: None ) –

  The labels of the graph object.

• properties (set[str] | None, default: None ) –

  The properties of the graph object.

Source code in gnfd/gnfd.py

class GraphObject(abc.ABC):
    """Abstract class representing an LPG graph object, i.e., a :any:`Edge` or a :any:`Node`.

    :param symbol: The symbol of the variable of the graph object.
    :type symbol: str
    :param labels: The labels of the graph object.
    :type labels: set[str] | None
    :param properties: The properties of the graph object.
    :type properties: set[str] | None"""

    def __init__(
        self,
        symbol: str = "",
        labels: set[str] | None = None,
        properties: set[str] | None = None,
    ):
        self.unique = uuid.uuid4().hex[:7]

        if symbol == "":
            self.symbol = "".join(random.choice(["a","b","c","d","e","f","g"]) for i in range(5))
        else:
            self.symbol: str = symbol



        if labels is None:
            self.labels: set[str] = set()
        else:
            self.labels: set[str] = labels

        if properties is None:
            self.properties: set[str] = set()
        else:
            self.properties: set[str] = properties

    @property
    def _gnfd_middle(self):
        if len(self.labels) < 1 and len(self.properties) < 1:
            middle = self.symbol
        elif len(self.properties) < 1:
            middle = f"{self.symbol}:{"&".join(self.labels)}"
        else:
            middle = f"{self.symbol}:{"&".join(self.labels)}:{"&".join(self.properties)}"
        return middle

    def clingo_symbol(self) -> str:
        """Returns a symbol for use with clingo."""


        clingo_symbol = f"{self.symbol}_{self.unique}"
        return clingo_symbol

    def is_anonymous(self):
        """A graph object is anonymous if it has no symbol.
        Anonymous graph objects cannot be used in the descriptor, i.e., the left and right side, of a :any:`GNFD`.
        """
        return self.symbol == ""

    @abstractmethod
    def go_equals(self, other: GraphObject):
        """:returns: Whether 2 graph objects are equivalent"""

    def __str__(self):
        if len(self.labels) < 1 and len(self.properties) < 1:
            return self.symbol
        elif len(self.properties) < 1:
            return f"{self.symbol}:{"6".join(self.labels)}"
        else:
            return f"{self.symbol}:{"6".join(self.labels)}:{",".join(self.properties)}"

clingo_symbol() -> str

Returns a symbol for use with clingo.

Source code in gnfd/gnfd.py

def clingo_symbol(self) -> str:
    """Returns a symbol for use with clingo."""


    clingo_symbol = f"{self.symbol}_{self.unique}"
    return clingo_symbol

go_equals(other: GraphObject) abstractmethod

Returns:

• –

  Whether 2 graph objects are equivalent

Source code in gnfd/gnfd.py

@abstractmethod
def go_equals(self, other: GraphObject):
    """:returns: Whether 2 graph objects are equivalent"""

is_anonymous()

A graph object is anonymous if it has no symbol. Anonymous graph objects cannot be used in the descriptor, i.e., the left and right side, of a :any:GNFD.

Source code in gnfd/gnfd.py

def is_anonymous(self):
    """A graph object is anonymous if it has no symbol.
    Anonymous graph objects cannot be used in the descriptor, i.e., the left and right side, of a :any:`GNFD`.
    """
    return self.symbol == ""

LeftEdge

Bases: Edge

Represents an LPG edge directed to the left.

Source code in gnfd/gnfd.py

class LeftEdge(Edge):
    """Represents an LPG edge directed to the left."""

    @property
    def leftmost_node(self):
        return self.tgt_pattern.leftmost_node

    @property
    def rightmost_node(self):
        return self.src_pattern.rightmost_node

    @property
    def src(self):
        return self.src_pattern.leftmost_node

    @property
    def tgt(self):
        return self.tgt_pattern.rightmost_node

    def minimal_pattern_intersections(self, other: Pattern):
        pass

    def __str__(self):
        return f"{str(self.tgt_pattern)}<-[{self._gnfd_middle}]-{str(self.src_pattern)}"

Node

Bases: GraphObject, Pattern

Represents an LPG node.

Source code in gnfd/gnfd.py

class Node(GraphObject, Pattern):
    """Represents an LPG node."""

    def get_graph_objects(self) -> set[GraphObject]:
        res = set()
        res.add(self)
        return res

    def get_graph_object_with_symbol(self, symbol: str) -> GraphObject | None:
        if self.symbol == symbol and symbol != "":
            return self
        else:
            return None

    def to_gql_match_where_tuple(self) -> tuple[str, str]:
        label_delim: str = ""
        if len(self.labels) > 0:
            label_delim = ":"
        return f"({self.symbol}{label_delim}{":".join(self.labels)})", f"{" AND ".join(map(lambda key: f"{self.symbol}.{key} IS NOT NULL", self.properties))}"

    def to_clingo_tableau(self, lhs, constants = True) -> str:
        res = ""
        node_symbol = self.clingo_symbol()

        def get_symbol(prop_symbol, suffix="") -> str:
            if f"{self.symbol}.{prop_symbol}" in map(str, lhs):
                return f"{self.symbol}_{prop_symbol}"
            else:
                return f"{self.symbol}_{prop_symbol}_{suffix}"

        for i in ["1","2"]:
            if self not in map(lambda ref: ref.reference, lhs):
                node_symbol += i  # the symbol of the node occurs not on the left side of a dep, so it must be different
            if constants:
                res += "".join(
                    map(
                        lambda lab: f"enc({node_symbol.lower()},lab__,str__{lab}, {uuid.uuid4().hex[:7].lower()}).\n",
                        self.labels,
                    )
                )
                res += "".join(
                    map(
                        lambda prop: f"enc({node_symbol.lower()},{get_symbol(prop)}__,str__{get_symbol(prop,i)}, {uuid.uuid4().hex[:7].lower()}).\n",
                        self.properties,
                    )
                )
            else:
                if constants:
                    res += "".join(
                        map(
                            lambda lab: f"enc({node_symbol.upper()},lab__,Str__{lab}, {uuid.uuid4().hex[:7].upper()}).\n",
                            self.labels,
                        )
                    )
                    res += "".join(
                        map(
                            lambda
                                prop: f"enc({node_symbol.upper()},{get_symbol(prop)}__,Str__{get_symbol(prop,i)}, {uuid.uuid4().hex[:7].upper()}).\n",
                            self.properties,
                        )
                    )
        return res

    def contains_var(self, symbol: str) -> bool:
        return symbol == self.symbol

    def contains_property(self, symbol: str, key: str):
        return self.contains_var(symbol) and key in self.properties

    def less_than(self, other):
        if not isinstance(other, Node):
            return False
        return ((other.labels.issubset(self.labels) and other.properties.issubset(self.properties)) or
                (other.labels == self.labels and other.properties.issubset(self.properties)) or
        (other.labels.issubset(self.labels) and other.properties == self.properties))

    def minimal_pattern_intersections(self, other: Pattern):
        res: list[tuple[Pattern, set[tuple[str, str]]]] = []

        if isinstance(other, Node): # case 1
            rep = set()
            rep.add((self.symbol, other.symbol))
            tup = (Node(self.symbol,
                        labels=self.labels.union(other.labels),
                        properties=self.properties.union(other.properties)), rep)
            res.append(tup)

        if isinstance(other, Edge): # case 2
            for mpi in self.minimal_pattern_intersections(other.src_pattern): # case 2.1
                other_clone = copy.deepcopy(other)
                other_clone.src_pattern = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)
            for mpi in self.minimal_pattern_intersections(other.tgt_pattern): # case 2.1
                other_clone = copy.deepcopy(other)
                other_clone.tgt_pattern = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)

        if isinstance(other, PatternConcat): # case 3
            for mpi in self.minimal_pattern_intersections(other.left): # case 3.1
                other_clone = copy.deepcopy(other)
                other_clone.left = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)
            for mpi in self.minimal_pattern_intersections(other.right): # case 3.2
                other_clone = copy.deepcopy(other)
                other_clone.right = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)
        return res

    def go_equals(self, other):
        return (
                isinstance(other, Node)
                and self.labels == other.labels
                and self.properties == other.properties
        )

    def equals(self, other: Pattern) -> bool:
        return (
            isinstance(other, Node)
            and self.labels == other.labels
            and self.properties == other.properties
        )




    @property
    def leftmost_node(self):
        return self

    @property
    def rightmost_node(self):
        return self

    def __str__(self):
        return f"({self._gnfd_middle})"

Pattern

Bases: ABC

Represents a pattern used for matching in an LPG.

Source code in gnfd/gnfd.py

class Pattern(abc.ABC):
    """Represents a pattern used for matching in an LPG."""

    def __init__(self, string: str):
        self.string: str = string

    @abstractmethod
    def to_clingo_tableau(self, lhs: set[Reference], constants: bool = True) -> str:
        """:returns: The pattern represented as clingo facts containing constants or variables."""
        pass

    @abstractmethod
    def to_gql_match_where_tuple(self) -> tuple[str, str]:
        """:returns: A tuple where the first value denotes a GQL match and the second denotes a GQL where"""
        pass

    def to_gql_match_where_string(self) -> str:
        tup = self.to_gql_match_where_tuple()
        if len(tup[1]) > 0: # WHERE followed with no statements would be illegal!
            return f"MATCH {tup[0]} WHERE {tup[1]}"
        else:
            return f"MATCH {tup[0]}"

    @abstractmethod
    def contains_var(self, symbol: str) -> bool:
        """Checks whether the pattern contains a graph object whose variable symbol is :paramref:`symbol`.

        :param symbol: The symbol which should be looked for in the pattern.
        :type symbol: str
        :returns: Whether there is a :any:`GraphObject` having :paramref:`symbol` as its symbol.
        """
        pass

    @abstractmethod
    def contains_property(self, symbol: str, key: str):
        """Checks whether the pattern contains a graph object whose variable symbol is :paramref:`symbol` and which has a property :paramref:`key`.

        :param symbol: The symbol which should be looked for in the pattern.
        :type symbol: str
        :param key: The key of the property whose existence should be checked.
        :type key: str
        :returns: Whether there is a :any:`GraphObject` having :paramref:`symbol` as its symbol.
        """
        pass

    @abstractmethod
    def get_graph_object_with_symbol(self, symbol: str) -> GraphObject | None:
        """:returns: a graph object with the given symbol"""
        pass

    @property
    @abc.abstractmethod
    def leftmost_node(self):
        """:returns: the left most node of the pattern, i.e., the leaf of always choosing pattern 1"""
        pass

    @property
    @abc.abstractmethod
    def rightmost_node(self):
        """:returns: the right most node of the pattern, i.e., the leaf of always choosing pattern 2"""
        pass

    def equals(self, other: Pattern) -> bool:
        """Computes whether the structure of this and the :paramref:`other` pattern and their labels and properties are equal.

        :returns: whether the :paramref:`other` is equal to this pattern."""
        for go1 in self.get_graph_objects():
            if not reduce(operator.or_, map(lambda go2: go1.go_equals(go2), other.get_graph_objects())):
                return False
        return True

    @abstractmethod
    def less_than(self, other) -> bool:
        """Computes whether the structure of this pattern is more specific than the other pattern."""
        pass

    @abstractmethod
    def minimal_pattern_intersections(
        self, other: Pattern
    ) -> list[tuple[Pattern, set[tuple[str, str]]]]:
        """Computes the minimal pattern intersection between this and the :paramref:`other` pattern.

        :returns: a list of tuples consisting of (1) the pattern that is the minimal intersection between this and the :paramref:`other` pattern and of (2) tuples denote variable correspondences between the patterns.
        """

    @abstractmethod
    def get_graph_objects(self) -> set[GraphObject]:
        """:returns: all graph objects, i.e., any :any:`Node` or :any:`Edge`."""
        pass

    def __str__(self):
        return self.string

leftmost_node abstractmethod property

Returns:

• –

  the left most node of the pattern, i.e., the leaf of always choosing pattern 1

rightmost_node abstractmethod property

Returns:

• –

  the right most node of the pattern, i.e., the leaf of always choosing pattern 2

contains_property(symbol: str, key: str) abstractmethod

Checks whether the pattern contains a graph object whose variable symbol is :paramref:symbol and which has a property :paramref:key.

Parameters:

• symbol (str) –

  The symbol which should be looked for in the pattern.

• key (str) –

  The key of the property whose existence should be checked.

Returns:

• –

  Whether there is a :any:GraphObject having :paramref:symbol as its symbol.

Source code in gnfd/gnfd.py

@abstractmethod
def contains_property(self, symbol: str, key: str):
    """Checks whether the pattern contains a graph object whose variable symbol is :paramref:`symbol` and which has a property :paramref:`key`.

    :param symbol: The symbol which should be looked for in the pattern.
    :type symbol: str
    :param key: The key of the property whose existence should be checked.
    :type key: str
    :returns: Whether there is a :any:`GraphObject` having :paramref:`symbol` as its symbol.
    """
    pass

contains_var(symbol: str) -> bool abstractmethod

Checks whether the pattern contains a graph object whose variable symbol is :paramref:symbol.

Parameters:

• symbol (str) –

  The symbol which should be looked for in the pattern.

Returns:

• bool –

  Whether there is a :any:GraphObject having :paramref:symbol as its symbol.

Source code in gnfd/gnfd.py

@abstractmethod
def contains_var(self, symbol: str) -> bool:
    """Checks whether the pattern contains a graph object whose variable symbol is :paramref:`symbol`.

    :param symbol: The symbol which should be looked for in the pattern.
    :type symbol: str
    :returns: Whether there is a :any:`GraphObject` having :paramref:`symbol` as its symbol.
    """
    pass

equals(other: Pattern) -> bool

Computes whether the structure of this and the :paramref:other pattern and their labels and properties are equal.

Returns:

• bool –

  whether the :paramref:other is equal to this pattern.

Source code in gnfd/gnfd.py

def equals(self, other: Pattern) -> bool:
    """Computes whether the structure of this and the :paramref:`other` pattern and their labels and properties are equal.

    :returns: whether the :paramref:`other` is equal to this pattern."""
    for go1 in self.get_graph_objects():
        if not reduce(operator.or_, map(lambda go2: go1.go_equals(go2), other.get_graph_objects())):
            return False
    return True

get_graph_object_with_symbol(symbol: str) -> GraphObject | None abstractmethod

Returns:

• GraphObject | None –

  a graph object with the given symbol

Source code in gnfd/gnfd.py

@abstractmethod
def get_graph_object_with_symbol(self, symbol: str) -> GraphObject | None:
    """:returns: a graph object with the given symbol"""
    pass

get_graph_objects() -> set[GraphObject] abstractmethod

Returns:

• set[GraphObject] –

  all graph objects, i.e., any :any:Node or :any:Edge.

Source code in gnfd/gnfd.py

@abstractmethod
def get_graph_objects(self) -> set[GraphObject]:
    """:returns: all graph objects, i.e., any :any:`Node` or :any:`Edge`."""
    pass

less_than(other) -> bool abstractmethod

Computes whether the structure of this pattern is more specific than the other pattern.

Source code in gnfd/gnfd.py

@abstractmethod
def less_than(self, other) -> bool:
    """Computes whether the structure of this pattern is more specific than the other pattern."""
    pass

minimal_pattern_intersections(other: Pattern) -> list[tuple[Pattern, set[tuple[str, str]]]] abstractmethod

Computes the minimal pattern intersection between this and the :paramref:other pattern.

Returns:

• list[tuple[Pattern, set[tuple[str, str]]]] –

  a list of tuples consisting of (1) the pattern that is the minimal intersection between this and the :paramref:other pattern and of (2) tuples denote variable correspondences between the patterns.

Source code in gnfd/gnfd.py

@abstractmethod
def minimal_pattern_intersections(
    self, other: Pattern
) -> list[tuple[Pattern, set[tuple[str, str]]]]:
    """Computes the minimal pattern intersection between this and the :paramref:`other` pattern.

    :returns: a list of tuples consisting of (1) the pattern that is the minimal intersection between this and the :paramref:`other` pattern and of (2) tuples denote variable correspondences between the patterns.
    """

to_clingo_tableau(lhs: set[Reference], constants: bool = True) -> str abstractmethod

Returns:

• str –

  The pattern represented as clingo facts containing constants or variables.

Source code in gnfd/gnfd.py

@abstractmethod
def to_clingo_tableau(self, lhs: set[Reference], constants: bool = True) -> str:
    """:returns: The pattern represented as clingo facts containing constants or variables."""
    pass

to_gql_match_where_tuple() -> tuple[str, str] abstractmethod

Returns:

• tuple[str, str] –

  A tuple where the first value denotes a GQL match and the second denotes a GQL where

Source code in gnfd/gnfd.py

@abstractmethod
def to_gql_match_where_tuple(self) -> tuple[str, str]:
    """:returns: A tuple where the first value denotes a GQL match and the second denotes a GQL where"""
    pass

PatternConcat

Bases: Pattern

Represents a concatenation of two patterns.

Source code in gnfd/gnfd.py

class PatternConcat(Pattern):
    """Represents a concatenation of two patterns."""

    def get_graph_objects(self) -> set[GraphObject]:
        res = set()
        res = res.union(self.left.get_graph_objects())
        res = res.union(self.right.get_graph_objects())
        return res

    def to_clingo_tableau(self) -> str:
        return f"{self.left.to_clingo_tableau()}\n{self.right.to_clingo_tableau()}"

    def __init__(self, left: Pattern, right: Pattern, string: str):
        super().__init__(string)
        self.left: Pattern = left
        self.right: Pattern = right

    def to_gql_match_where_tuple(self) -> tuple[str, str]:
        tup_left_pattern = self.left.to_gql_match_where_tuple()
        tup_right_pattern = self.right.to_gql_match_where_tuple()

        # Don't match with every node! --> would result in a match with the cartesian product
        if tup_left_pattern[0] == "()":
            return tup_right_pattern
        elif tup_right_pattern[0] == "()":
            return tup_left_pattern

        where = tup_left_pattern[1]
        if len(where) > 0 and len(tup_right_pattern[1]) > 0:
            where += " AND "
        where += tup_right_pattern[1]

        return f"{tup_left_pattern[0]},{tup_right_pattern[0]}", where

    def contains_var(self, symbol: str) -> bool:
        return self.left.contains_var(symbol) or self.right.contains_var(symbol)

    def contains_property(self, symbol: str, key: str):
        return self.left.contains_property(symbol, key) or self.right.contains_property(
            symbol, key
        )

    def get_graph_object_with_symbol(self, symbol: str) -> GraphObject | None:
        lgo = self.left.get_graph_object_with_symbol(symbol)
        rgo = self.right.get_graph_object_with_symbol(symbol)
        if lgo is not None:
            return lgo
        else:
            return rgo

    @property
    def leftmost_node(self):
        return self.left.leftmost_node

    @property
    def rightmost_node(self):
        return self.right.rightmost_node

    def less_than(self, other):
        if not isinstance(other, PatternConcat):
            return self.left.less_than(other) or self.right.less_than(other)
        else:
            return ((self.left.less_than(other.left) and self.right.less_than(other.right)) or
        (self.left.less_than(other.right) and self.right.less_than(other.left)) or
                    (self.left.equals(other.left) and self.right.less_than(other.right)) or
                    (self.left.equals(other.right) and self.right.less_than(other.left)) or
                    (self.left.less_than(other.left) and self.right.equals(other.right)) or
                    (self.left.less_than(other.right) and self.right.equals(other.left))
                    )

    def minimal_pattern_intersections(self, other: Pattern):
        if isinstance(other, Node) or isinstance(other, Edge):
            return other.minimal_pattern_intersections(self)

        else: # case 6
            assert isinstance(other, PatternConcat)

            res: list[tuple[Pattern, set[tuple[str, str]]]] = []

            for mpi in self.minimal_pattern_intersections(other.left): # case 6.1
                other_clone = copy.deepcopy(other)
                other_clone.left = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)
            for mpi in self.minimal_pattern_intersections(other.right): # case 6.2
                other_clone = copy.deepcopy(other)
                other_clone.left = mpi[0]
                rep = set()
                rep = rep.union(mpi[1])
                tup = (other_clone, rep)
                res.append(tup)
            for mpi1 in self.minimal_pattern_intersections(other.left): # case 6.3
                for mpi2 in self.minimal_pattern_intersections(other.right):
                    other_clone = copy.deepcopy(other)
                    other_clone.left = mpi1[0]
                    other_clone.right = mpi2[0]
                    rep = set()
                    rep = rep.union(mpi1[1])
                    rep = rep.union(mpi2[1])

                    tup = (other_clone, rep)
                    res.append(tup)

            return res

    def __str__(self):
        return f"{str(self.left)}, {str(self.right)}"

Property

Represents an LPG property, i.e., a combination of property key and graph object variable.

Source code in gnfd/gnfd.py

class Property:
    """Represents an LPG property, i.e., a combination of property key and graph object variable."""

    def __init__(self, key: str, graph_object: GraphObject):
        self.key: str = key
        self.graphObject: GraphObject = graph_object

    def equals(self, other: Property) -> bool:
        return self.graphObject.go_equals(other.graphObject) and self.key == other.key

    def __str__(self):
        return f"{str(self.graphObject.symbol)}.{self.key}"

Reference

Represents a reference to a property key contained in a graph object or to the graph object itself.

Source code in gnfd/gnfd.py

class Reference:
    """Represents a reference to a property key contained in a graph object or to the graph object itself."""

    def __init__(self, reference: GraphObject | Property):
        self.reference: GraphObject | Property = reference

    def __str__(self):
        if isinstance(self.reference, GraphObject):
            return self.reference.symbol
        return f"{str(self.reference)}"

    def get_graph_object(self) -> GraphObject:
        """:return: The graph object of the reference."""
        if isinstance(self.reference, GraphObject):
            return self.reference
        else:
            assert isinstance(self.reference, Property)
            return self.reference.graphObject

    def equals(self, other: Reference) -> bool:
        if isinstance(self.reference, GraphObject) and isinstance(other.reference, GraphObject):
            return self.reference.go_equals(other.reference)
        elif isinstance(self.reference, Property) and isinstance(self.reference, Property):
            return self.reference.equals(other.reference)
        else:
            return False # A graph object reference and a property reference can never be equal

    def to_query_string(self, database: str) -> str:
        """:return: The string representation of the reference that can be used to query the provided database."""
        if database == "memgraph" and isinstance(self.reference, GraphObject):
            return f"id({str(self.reference.symbol)})"
        elif isinstance(self.reference, GraphObject):
            return f"elementId({self.reference.symbol})"
        else:
            return str(self)

    @property
    def is_property_variable(self):
        return isinstance(self.reference, Property)

    @property
    def is_graph_object_variable(self):
        return isinstance(self.reference, GraphObject)

get_graph_object() -> GraphObject

Returns:

• GraphObject –

  The graph object of the reference.

Source code in gnfd/gnfd.py

def get_graph_object(self) -> GraphObject:
    """:return: The graph object of the reference."""
    if isinstance(self.reference, GraphObject):
        return self.reference
    else:
        assert isinstance(self.reference, Property)
        return self.reference.graphObject

to_query_string(database: str) -> str

Returns:

• str –

  The string representation of the reference that can be used to query the provided database.

Source code in gnfd/gnfd.py

def to_query_string(self, database: str) -> str:
    """:return: The string representation of the reference that can be used to query the provided database."""
    if database == "memgraph" and isinstance(self.reference, GraphObject):
        return f"id({str(self.reference.symbol)})"
    elif isinstance(self.reference, GraphObject):
        return f"elementId({self.reference.symbol})"
    else:
        return str(self)

RightEdge

Bases: Edge

Represents an LPG edge directed to the right.

Source code in gnfd/gnfd.py

class RightEdge(Edge):
    """Represents an LPG edge directed to the right."""

    @property
    def leftmost_node(self):
        return self.src_pattern.leftmost_node

    @property
    def rightmost_node(self):
        return self.tgt_pattern.rightmost_node

    @property
    def src(self):
        return self.src_pattern.rightmost_node

    @property
    def tgt(self):
        return self.tgt_pattern.leftmost_node

    def minimal_pattern_intersections(self, other: Pattern):
        pass

    def __str__(self):
        return f"{str(self.src_pattern)}-[{self._gnfd_middle}]->{str(self.tgt_pattern)}"

---

1. Philipp Skavantzos and Sebastian Link. 2025. Third and Boyce-Codd normal form for property graphs. VLDB J. 34, 2 (2025), 23. ↩