Keep track of reverse-cardinality in Joins for optimizations (#413)

Modifying the cardinality set up for relational join nodes to have a notion of "reverse cardinality", e.g. what is the cardinality from the perspective of the RHS input with regards to the LHS input. For example, when the left hand side is the join is the tpch CUSTOMER table and the right hand side is the tpch ORDERS table, the cardinality is `PLURAL_FILTER` (since each customer can have 0, 1, or multiple matching orders) but the reverse cardinality is `SINGULAR_ACCESS` (since each order has exactly 1 matching customer). This reverse cardinality can be used for two things right away:
- Adjusting the partial aggregation splitting protocol to infer when to push / not push an aggregate into the _right_ hand side based on the reverse cardinality (e.g. if the reverse cardinality is filtering, don't push because the join will actually reduce the number of rows).
- Modifying the column pruning protocol for joins, which currently removes the RHS for certain kinds of joins if the RHS columns are unused and the cardinality is `SINGULAR_ACCESS`: can now dot he same to prune the LHS entirely if the reverse cardinality is `SINGULAR_ACCESS` (e.g. in the `CUSTOMER` -> `ORDERS` example, if it is an inner join and every column in `CUSTOMER` is unused, we can just prune the `CUSTOMER` side of the join entirely)

Author: knassre-bodo

demos/metadata/tpch_demo_graph.json

@@ -7,7 +7,7 @@
         "name": "regions",
         "type": "simple table",
         "table path": "main.REGION",
-        "unique properties": ["key"],
+        "unique properties": ["key", "name"],
         "properties": [
           {
             "name": "key",
@@ -41,7 +41,7 @@
         "name": "nations",
         "type": "simple table",
         "table path": "main.NATION",
-        "unique properties": ["key"],
+        "unique properties": ["key", "name"],
         "properties": [
           {
             "name": "key",
@@ -83,7 +83,7 @@
         "name": "parts",
         "type": "simple table",
         "table path": "main.PART",
-        "unique properties": ["key"],
+        "unique properties": ["key", "name"],
         "properties": [
           {
             "name": "key",
@@ -170,7 +170,7 @@
         "name": "suppliers",
         "type": "simple table",
         "table path": "main.SUPPLIER",
-        "unique properties": ["key", "name"],
+        "unique properties": ["key", "name", "phone", "address"],
         "properties": [
           {
             "name": "key",
@@ -527,7 +527,7 @@
         "name": "customers",
         "type": "simple table",
         "table path": "main.CUSTOMER",
-        "unique properties": ["key", "name"],
+        "unique properties": ["key", "name", "address"],
         "properties": [
           {
             "name": "key",
@@ -813,6 +813,9 @@
         "description": "The orders that a customer has placed, each of which contains one or more line items",
         "synonyms": ["transactions", "purchases"]
       }
-    ]
-  }
+    ],
+    "additional definitions": [],
+    "verified pydough analysis": [],
+    "extra semantic info": {}
+  },
 ]

pydough/conversion/agg_split.py

@@ -355,6 +355,7 @@ def attempt_join_aggregate_transpose(
     # if joining first will reduce the number of rows that get aggregated.
     if join.cardinality.filters:
         can_push_left = False
+    if join.reverse_cardinality.filters:
         can_push_right = False
 
     # If any of the aggregations to either side cannot be pushed down, then
@@ -468,6 +469,9 @@ def attempt_join_aggregate_transpose(
             )
             node.aggregations[count_call_name] = regular_sum
             node.columns[count_call_name] = regular_sum
+        projection_columns[count_call_name] = ColumnReference(
+            count_call_name, NumericType()
+        )
 
     # If the node requires projection at the end, create a new Project node on
     # top of the top aggregate.

pydough/conversion/filter_pushdown.py

@@ -186,6 +186,7 @@ def visit_join(self, join: Join) -> RelationalNode:
         # The join type, cardinality, and inputs for the output join node.
         join_type: JoinType = join.join_type
         cardinality: JoinCardinality = join.cardinality
+        reverse_cardinality: JoinCardinality = join.reverse_cardinality
         new_inputs: list[RelationalNode] = []
 
         # If the join type is LEFT or SEMI but the condition is TRUE, convert it
@@ -240,10 +241,14 @@ def visit_join(self, join: Join) -> RelationalNode:
                     remaining_filters,
                     lambda expr: only_references_columns(expr, input_cols[idx]),
                 )
-            # Ensure that if any filter is pushed into an input (besides
-            # the first input) that the join is marked as filtering.
-            if len(pushable_filters) > 0 and idx > 0:
-                cardinality = join.cardinality.add_filter()
+            # Ensure that if any filter is pushed into an input, the
+            # corresponding join cardinality is updated to reflect that a filter
+            # has been applied.
+            if len(pushable_filters) > 0:
+                if idx == 1:
+                    cardinality = join.cardinality.add_filter()
+                else:
+                    reverse_cardinality = reverse_cardinality.add_filter()
             pushable_filters = {
                 expr.accept_shuttle(transposer) for expr in pushable_filters
             }
@@ -271,6 +276,7 @@ def visit_join(self, join: Join) -> RelationalNode:
             else:
                 new_conjunction.add(join._condition)
             cardinality = join.cardinality.add_filter()
+            reverse_cardinality = join.reverse_cardinality.add_filter()
             join._condition = RelationalExpression.form_conjunction(
                 sorted(new_conjunction, key=repr)
             )
@@ -281,6 +287,7 @@ def visit_join(self, join: Join) -> RelationalNode:
         new_node = join.copy(inputs=new_inputs)
         assert isinstance(new_node, Join)
         new_node.cardinality = cardinality
+        new_node.reverse_cardinality = reverse_cardinality
         new_node.join_type = join_type
         return build_filter(new_node, remaining_filters)
 

pydough/conversion/hybrid_connection.py

@@ -10,7 +10,7 @@
 from enum import Enum
 from typing import TYPE_CHECKING
 
-from pydough.relational import JoinType
+from pydough.relational import JoinCardinality, JoinType
 
 from .hybrid_expressions import (
     HybridFunctionExpr,
@@ -313,6 +313,8 @@ class HybridConnection:
        child can be defined at (exclusive).
     - `aggs`: a mapping of aggregation calls made onto expressions relative to the
        context of `subtree`.
+    - `reverse_cardinality`: the JoinCardinality of the connection from the
+       perspective of the child subtree back to the parent tree.
     """
 
     parent: "HybridTree"
@@ -349,6 +351,12 @@ class HybridConnection:
     expressions defined relative to the child subtree.
     """
 
+    reverse_cardinality: JoinCardinality
+    """
+    The JoinCardinality of the connection from the perspective of the child
+    subtree back to the parent tree.
+    """
+
     always_exists: bool | None = None
     """
     Whether the connection is guaranteed to have at least one matching

pydough/conversion/hybrid_correlation_extraction.py

@@ -238,6 +238,9 @@ def attempt_correlation_extraction(
                         for _, rhs_key in new_equi_filters:
                             bottom_subtree.agg_keys.append(rhs_key)
                     connection.always_exists = False
+                    connection.reverse_cardinality = (
+                        connection.reverse_cardinality.add_filter()
+                    )
 
                 if len(new_general_filters) > 0:
                     if bottom_subtree.general_join_condition is not None:
@@ -262,6 +265,9 @@ def attempt_correlation_extraction(
                             pydop.BAN, new_general_filters, BooleanType()
                         )
                     connection.always_exists = False
+                    connection.reverse_cardinality = (
+                        connection.reverse_cardinality.add_filter()
+                    )
 
                 # Update the filter condition with the new conjunction of terms
                 if new_conjunction != conjunction:

pydough/conversion/hybrid_decorrelater.py

@@ -9,6 +9,7 @@
 import copy
 
 import pydough.pydough_operators as pydop
+from pydough.relational import JoinCardinality
 from pydough.types import BooleanType
 
 from .hybrid_connection import ConnectionType, HybridConnection
@@ -430,6 +431,13 @@ def decorrelate_child(
         )
         if child.connection_type.is_aggregation or is_faux_agg:
             child.subtree.agg_keys = new_agg_keys
+
+        # Mark the reverse cardinality as SINGULAR_ACCESS since each record of
+        # the de-correlated child can only match with one record of the
+        # original parent due to the join keys being based on the uniqueness
+        # keys of the original parent.
+        child.reverse_cardinality = JoinCardinality.SINGULAR_ACCESS
+
         # If the child is such that we don't need to keep rows from the parent
         # without a match, replace the parent & its ancestors with a
         # HybridPullUp node (and replace any other deleted nodes with no-ops).

pydough/conversion/hybrid_tree.py

@@ -18,11 +18,13 @@
 from pydough.metadata import (
     SubcollectionRelationshipMetadata,
 )
+from pydough.metadata.properties import ReversiblePropertyMetadata
 from pydough.qdag import (
     Literal,
     SubCollection,
     TableCollection,
 )
+from pydough.relational import JoinCardinality
 from pydough.types import BooleanType, NumericType
 
 from .hybrid_connection import ConnectionType, HybridConnection
@@ -579,13 +581,29 @@ def add_child(
                 # Return the index of the existing child.
                 return idx
 
+        # Infer the cardinality of the join from the perspective of the new
+        # collection to the existing data.
+        reverse_cardinality: JoinCardinality = child.infer_root_reverse_cardinality(
+            self
+        )
+
         # Create and insert the new child connection.
         new_child_idx = len(self.children)
         connection: HybridConnection = HybridConnection(
-            self, child, connection_type, min_steps, max_steps, {}
+            self,
+            child,
+            connection_type,
+            min_steps,
+            max_steps,
+            {},
+            reverse_cardinality,
         )
         self._children.append(connection)
 
+        # Augment the reverse cardinality if the parent does not always exist.
+        if (not reverse_cardinality.filters) and (not self.always_exists()):
+            connection.reverse_cardinality = reverse_cardinality.add_filter()
+
         # If an operation prevents the child's presence from directly
         # filtering the current level, update its connection type to be either
         # SINGULAR or AGGREGATION, then insert a similar COUNT(*)/PRESENT
@@ -605,6 +623,96 @@ def add_child(
         # Return the index of the newly created child.
         return new_child_idx
 
+    @staticmethod
+    def infer_metadata_reverse_cardinality(
+        metadata: SubcollectionRelationshipMetadata,
+    ) -> JoinCardinality:
+        """
+        Infers the cardinality of the reverse of a join (child → parent)
+        based on the metadata of the reverse-relationship, if one exists.
+        If no reverse metadata exists, defaults to PLURAL_FILTER (safest assumption)
+
+        Args:
+            `metadata`: the metadata for the sub-collection property mapping
+            the parent to the child.
+
+        Returns:
+            The join cardinality for the connection from the child back to the
+            parent, if it can be inferred. Uses `PLURAL_FILTER` as a fallback.
+        """
+        # If there is no reverse, fall back to plural filter (which is the
+        # safest default assumption).
+        if (
+            not isinstance(metadata, ReversiblePropertyMetadata)
+            or metadata.reverse is None
+        ):
+            return JoinCardinality.PLURAL_FILTER
+
+        # If the reverse property exists, use its properties to
+        # infer if the reverse cardinality is singular or plural
+        # and whether a match always exists or not.
+        cardinality: JoinCardinality
+        match (metadata.reverse.is_plural, metadata.reverse.always_matches):
+            case (False, True):
+                cardinality = JoinCardinality.SINGULAR_ACCESS
+            case (False, False):
+                cardinality = JoinCardinality.SINGULAR_FILTER
+            case (True, True):
+                cardinality = JoinCardinality.PLURAL_ACCESS
+            case (True, False):
+                cardinality = JoinCardinality.PLURAL_FILTER
+        return cardinality
+
+    def infer_root_reverse_cardinality(self, context: "HybridTree") -> JoinCardinality:
+        """
+        Infers the cardinality of the join connecting the root of the hybrid
+        tree back to its parent context.
+
+        Args:
+            `context`: the parent context that the root of the hybrid tree is
+            being connected to.
+
+        Returns:
+            The inferred cardinality of the join connecting the root of the
+            hybrid tree to its parent context.
+        """
+        # Keep traversing upward until we find the root of the current tree.
+        if self.parent is not None:
+            return self.parent.infer_root_reverse_cardinality(context)
+
+        # Once we find the root, infer the cardinality of the join that would
+        # connect just this node to the parent context.
+        # At the root, only this node’s type matters for reverse cardinality.
+        # Deeper nodes do not affect parent-child match guarantees.
+        match self.pipeline[0]:
+            case HybridRoot():
+                # If the parent of the child is a root, it means a cross join
+                # is occurring, so the cardinality depends on whether
+                # the parent context is singular or plural.
+                return (
+                    JoinCardinality.SINGULAR_ACCESS
+                    if context.is_singular()
+                    else JoinCardinality.PLURAL_ACCESS
+                )
+            case HybridCollectionAccess():
+                # For non sub-collection accesses, use plural access.
+                # For a sub-collection, infer from the reverse property.
+                if isinstance(self.pipeline[0].collection, SubCollection):
+                    return self.infer_metadata_reverse_cardinality(
+                        self.pipeline[0].collection.subcollection_property
+                    )
+                else:
+                    return JoinCardinality.PLURAL_ACCESS
+            # For partition & partition child, infer from the underlying child.
+            case HybridPartition():
+                return self.children[0].subtree.infer_root_reverse_cardinality(context)
+            case HybridPartitionChild():
+                return self.pipeline[0].subtree.infer_root_reverse_cardinality(context)
+            case _:
+                raise NotImplementedError(
+                    f"Invalid start of pipeline: {self.pipeline[0].__class__.__name__}"
+                )
+
     def add_successor(self, successor: "HybridTree") -> None:
         """
         Marks two hybrid trees in a predecessor-successor relationship.
@@ -723,7 +831,7 @@ def is_singular(self) -> bool:
         match self.pipeline[0]:
             case HybridCollectionAccess():
                 if isinstance(self.pipeline[0].collection, TableCollection):
-                    pass
+                    return False
                 else:
                     assert isinstance(self.pipeline[0].collection, SubCollection)
                     meta: SubcollectionRelationshipMetadata = self.pipeline[
@@ -734,6 +842,8 @@ def is_singular(self) -> bool:
             case HybridChildPullUp():
                 if not self.children[self.pipeline[0].child_idx].subtree.is_singular():
                     return False
+            case HybridRoot():
+                pass
             case _:
                 return False
         # The current level is fine, so check any levels above it next.