fix: bind a RIGHT join's merge arguments to the declared sides

docs/docs/in_depth/join_data.md

@@ -51,7 +51,7 @@ Join Types specify how two datasets are merged based on their keys. The framewor
 -   Inner Join,
 -   Left Join,
 -   Outer Join,
--   Right Join (use sparingly; prefer left joins when possible).
+-   Right Join (use sparingly; prefer left joins when possible). The declared left feature group's data is the merge engine's left argument, whichever compute framework the join executes in.
 -   ASOF Join (point-in-time / as-of: equi match on the by-keys, nearest time match on the time columns).
 
 ```python

mloda/core/prepare/execution_plan.py

@@ -17,7 +17,7 @@
 from mloda.core.prepare.joinstep_collection import JoinStepCollection
 from mloda.core.prepare.graph.graph import Graph
 from mloda.core.prepare.resolve_graph import PlannedQueue
-from mloda.core.prepare.resolve_links import LinkFrameworkTrekker, LinkTrekker
+from mloda.core.prepare.resolve_links import LinkFrameworkTrekker, LinkTrekker, inheritance_distance
 from mloda.core.core.step.feature_group_step import FeatureGroupStep
 from mloda.core.core.step.join_step import JoinStep
 from mloda.core.core.step.transform_frame_work_step import TransformFrameworkStep
@@ -42,6 +42,13 @@ def _filter_options_sort_key(single_filter: SingleFilter) -> tuple[str, str]:
     )
 
 
+def _nearest_frameworks(frameworks_by_distance: dict[int, set[type[ComputeFramework]]]) -> set[type[ComputeFramework]]:
+    """A declared side is held by its closest subclasses only; farther ones answer for a different side."""
+    if not frameworks_by_distance:
+        return set()
+    return frameworks_by_distance[min(frameworks_by_distance)]
+
+
 class ExecutionPlan:
     def __init__(
         self,
@@ -439,13 +446,31 @@ def run_link(
         destination_framework_uuids: set[UUID] = set()
         source_framework_uuids: set[UUID] = set()
 
+        left_frameworks_by_distance: dict[int, set[type[ComputeFramework]]] = defaultdict(set)
+        right_frameworks_by_distance: dict[int, set[type[ComputeFramework]]] = defaultdict(set)
+
         for uuid in required_uuids:
-            if graph.get_nodes()[uuid].feature.get_compute_framework() == destination_framework:
+            node = graph.get_nodes()[uuid]
+            node_framework = node.feature.get_compute_framework()
+
+            if node_framework == destination_framework:
                 destination_framework_uuids.add(uuid)
 
-            if graph.get_nodes()[uuid].feature.get_compute_framework() == source_framework:
+            if node_framework == source_framework:
                 source_framework_uuids.add(uuid)
 
+            # Links match polymorphically, so a subclass of a declared side counts as that side, ranked by distance.
+            if issubclass(node.feature_group_class, link.left_feature_group):
+                left_distance = inheritance_distance(node.feature_group_class, link.left_feature_group)
+                left_frameworks_by_distance[left_distance].add(node_framework)
+
+            if issubclass(node.feature_group_class, link.right_feature_group):
+                right_distance = inheritance_distance(node.feature_group_class, link.right_feature_group)
+                right_frameworks_by_distance[right_distance].add(node_framework)
+
+        declared_left_frameworks = _nearest_frameworks(left_frameworks_by_distance)
+        declared_right_frameworks = _nearest_frameworks(right_frameworks_by_distance)
+
         # The order shows which items should be added first.
         # Thus, we need to make sure that higher ordered links are calculated first.
         for k, v in link_trekker.order.items():
@@ -483,13 +508,33 @@ def run_link(
                 required_uuids,
                 destination_framework_uuids,
                 source_framework_uuids,
-                swap_merge_sides,
+                self.swap_merge_sides_by_declared_side(
+                    destination_framework, declared_left_frameworks, declared_right_frameworks, swap_merge_sides
+                ),
             )
 
         # This makes sure that we do not write on the same datasets due to overlapping joins at once.
         self.joinstep_collection.add(js)
         return js
 
+    @staticmethod
+    def swap_merge_sides_by_declared_side(
+        destination_framework: type[ComputeFramework],
+        declared_left_frameworks: set[type[ComputeFramework]],
+        declared_right_frameworks: set[type[ComputeFramework]],
+        fallback: bool,
+    ) -> bool:
+        """The declared left group's data must stay the merge engine's left argument, wherever the join runs."""
+        holds_left = destination_framework in declared_left_frameworks
+        holds_right = destination_framework in declared_right_frameworks
+
+        if holds_left and not holds_right:
+            return False
+        if holds_right and not holds_left:
+            return True
+        # Self links and sides sharing one framework are not decidable from the declared sides.
+        return fallback
+
     def find_fg_per_uuid(
         self, pre_execution_plan: list[LinkFrameworkTrekker | FeatureGroupStep], uuid: UUID
     ) -> type[FeatureGroup]:

mloda/core/prepare/resolve_links.py

@@ -11,6 +11,12 @@
 LinkFrameworkTrekker = tuple[Link, type[ComputeFramework], type[ComputeFramework]]
 
 
+def inheritance_distance(child: type, parent: type) -> int:
+    """Steps from child to parent in the MRO, or 9999 if parent is not in child's hierarchy."""
+    mro: tuple[type, ...] = getattr(child, "__mro__", ())
+    return mro.index(parent) if parent in mro else 9999
+
+
 class LinkTrekker:
     """This class is used to keep track of Links and which children depend on this link."""
 
@@ -353,16 +359,7 @@ def _find_matching_links(
         return self._select_most_specific_links(polymorphic_matches, left_fg, right_fg)
 
     def _inheritance_distance(self, child: type, parent: type) -> int:
-        """Calculate the inheritance distance from child to parent in the MRO.
-
-        Returns the number of steps in the Method Resolution Order from child to parent.
-        Returns a large number if parent is not in child's MRO.
-        """
-        try:
-            mro = child.__mro__
-            return mro.index(parent)
-        except (ValueError, AttributeError):
-            return 9999  # Not in hierarchy
+        return inheritance_distance(child, parent)
 
     def _select_most_specific_links(self, links: list[Link], left_fg: type, right_fg: type) -> list[Link]:
         """Select links that are most specific (closest in inheritance hierarchy).

tests/test_core/test_integration/test_core/test_link_planner_orientation_characterization.py

@@ -293,11 +293,6 @@ def _run_pair(
 RIGHT_JOIN_ROWS = ["k3|L3|k3|R3", "k4|L4|k4|R4", f"{MISSING}|{MISSING}|k5|R5"]
 CHAIN_ROWS = ["k4|L4|R4|T4", "k3|L3|R3|T3"]
 
-RIGHT_SIDE_BINDING_REASON = (
-    "plain RIGHT joins bind the merge arguments to the resolved frameworks instead of the "
-    "declared sides, so the declared left index is looked up in the right group's data"
-)
-
 
 @MODES_WITH_MULTIPROCESSING
 def test_inner_join_declared_orientation_keeps_left_group_first(
@@ -348,17 +343,7 @@ def test_right_join_keeps_every_right_row_for_a_child_declaring_the_left_framewo
 
 
 @MODES_SYNC_THREADING
-def test_right_join_raises_for_a_child_on_the_right_framework(
-    modes: set[ParallelizationMode], flight_server: Any
-) -> None:
-    # The exception type is incidental: the column-semantics guard reaches the key column before the merge does.
-    with pytest.raises((KeyError, ValueError), match="oc_b_left_key"):
-        _run_pair(PAIR_B, "right", OrientCharInvertedChild, modes, flight_server)
-
-
-@pytest.mark.xfail(strict=True, reason=RIGHT_SIDE_BINDING_REASON)
-@MODES_SYNC_THREADING
-def test_right_join_should_keep_every_right_row_for_a_child_on_the_right_framework(
+def test_right_join_keeps_every_right_row_for_a_child_on_the_right_framework(
     modes: set[ParallelizationMode], flight_server: Any
 ) -> None:
     rows = _run_pair(PAIR_B, "right", OrientCharInvertedChild, modes, flight_server)