Extract correlated filters from the bottom of a hybrid tree into a join condition (#404)

Series of optimizations/adjustments  to hybrid trees and relational trees, largely to improve the final SQL:
- Runs a pass to extract as many correlated filter conditions as possible into the join keys/condition of the subtree. The pass runs right before de-correlation.
- Overrides unnest_subqueries.py from sqlglot with a customized version that avoids SQLGlot's de-correlation procedure when it would produce `ARRAY_AGG` + `ARRAY_ANY` solutions (caused via a ripple effect of the new pass)
- Adjusted de-correlation to preserve the join condition/keys from the now-replaced connection logic by adding them into the filters of the new subtree. This will ensure that any formerly correlated conditions that got pulled into the join get ejected back into the subtree as non-correlated filters if the subtree still gets de-correlated
- Added ability to remove redundant filters atop a join if they are already present inside a join
- Added the ability to push filters atop a join into the join condition unless there are window functions in the picture
- Adjusts how cardinality works with join filters to improve when joins are/aren't pruned, including accounting for filters that get pushed into a join's condition
- Added two new correl tests to account for some of the newer patterns (specifically with hybrid correlation extraction), and added all correl tests as SQL testss (only using sqlite), also deletes dead testing files.
- Splits up some of the logic of `pull_project_into_aggregate` to run even if the input to an aggregate is not a project (specifically the parts that simplify aggregate calls). Also, replaces casting booleans to integers in that helper with `IFF(x, 1, 0)`

Author: knassre-bodo

pydough/conversion/filter_pushdown.py

@@ -5,8 +5,10 @@
 __all__ = ["push_filters"]
 
 
+import pydough.pydough_operators as pydop
 from pydough.relational import (
     Aggregate,
+    CallExpression,
     ColumnReference,
     EmptySingleton,
     Filter,
@@ -240,6 +242,30 @@ def visit_join(self, join: Join) -> RelationalNode:
             self.filters = pushable_filters
             new_inputs.append(child.accept_shuttle(self))
 
+        # If there are no window functions in the filters, push any remaining
+        # filters into the join condition if it is an inner join.
+        if (
+            join_type == JoinType.INNER
+            and len(remaining_filters) > 0
+            and not any(contains_window(expr) for expr in remaining_filters)
+        ):
+            transposer.keep_input_names = True
+            new_conjunction: set[RelationalExpression] = set()
+            for expr in remaining_filters:
+                new_conjunction.add(expr.accept_shuttle(transposer))
+            if (
+                isinstance(join._condition, CallExpression)
+                and join._condition.op == pydop.BAN
+            ):
+                new_conjunction.update(join._condition.inputs)
+            else:
+                new_conjunction.add(join._condition)
+            cardinality = join.cardinality.add_filter()
+            join._condition = RelationalExpression.form_conjunction(
+                sorted(new_conjunction, key=repr)
+            )
+            remaining_filters = set()
+
         # Materialize all of the remaining filters on top of a new join with
         # the new inputs.
         new_node = join.copy(inputs=new_inputs)

pydough/conversion/hybrid_connection.py

@@ -316,11 +316,54 @@ class HybridConnection:
     """
 
     parent: "HybridTree"
+    """
+    The HybridTree that the connection exists within.
+    """
+
     subtree: "HybridTree"
+    """
+    The HybridTree corresponding to the child itself, starting from the bottom.
+    """
+
     connection_type: ConnectionType
+    """
+    The type of connection that the child subtree is being accessed with
+    relative to the parent hybrid tree.
+    """
+
     min_steps: int
+    """
+    The step in the parent pipeline that this connection MUST be defined after.
+    """
+
     max_steps: int
+    """
+    The step in the parent pipeline that this connection MUST be defined before.
+    """
+
     aggs: dict[str, HybridFunctionExpr]
+    """
+    A dictionary storing information about how to aggregate the child before
+    joining it to the parent. The keys are the names of the aggregation calls
+    within the child subtree, and the values are the corresponding aggregation
+    expressions defined relative to the child subtree.
+    """
+
+    always_exists: bool | None = None
+    """
+    Whether the connection is guaranteed to have at least one matching
+    record for every parent record. If None, this is unknown and must be
+    inferred from the subtree (can be stored and then modified later).
+    """
+
+    def get_always_exists(self) -> bool:
+        """
+        Returns whether the connection is guaranteed to have at least one
+        matching record for every parent record.
+        """
+        if self.always_exists is None:
+            self.always_exists = self.subtree.always_exists()
+        return self.always_exists
 
     def __eq__(self, other):
         return (