Rework join filter pushdown rewrites to leave hints and apply adaptively

python/cudf_polars/cudf_polars/dsl/ir.py

@@ -2543,7 +2543,12 @@ class Join(IR):
     """A join of two dataframes."""
 
     __slots__ = ("left_on", "options", "right_on")
-    _non_child = ("schema", "left_on", "right_on", "options")
+    _non_child: ClassVar[tuple[str, ...]] = (
+        "schema",
+        "left_on",
+        "right_on",
+        "options",
+    )
     _n_non_child_args = 3
     left_on: tuple[expr.NamedExpr, ...]
     """List of expressions used as keys in the left frame."""

python/cudf_polars/cudf_polars/dsl/utils/column_domain.py

@@ -21,6 +21,7 @@
     Slice,
     Sort,
 )
+from cudf_polars.streaming.filter_hint import PushdownFilterHint
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -141,3 +142,11 @@ def _(
     return {
         name: ColumnBinding(0, name) for name in node.schema if name in child.schema
     }
+
+
+@column_domain_bindings.register(PushdownFilterHint)
+def _(node: PushdownFilterHint) -> Mapping[str, ColumnBinding]:
+    target = node.children[0]
+    return {
+        name: ColumnBinding(0, name) for name in node.schema if name in target.schema
+    }

python/cudf_polars/cudf_polars/streaming/actor_graph/__init__.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 """RapidsMPF streaming-engine support."""
@@ -15,6 +15,7 @@
 import cudf_polars.streaming.actor_graph.io
 import cudf_polars.streaming.actor_graph.join
 import cudf_polars.streaming.actor_graph.over
+import cudf_polars.streaming.actor_graph.prefilter_actor
 import cudf_polars.streaming.actor_graph.repartition
 import cudf_polars.streaming.actor_graph.union  # noqa: F401
 

python/cudf_polars/cudf_polars/streaming/actor_graph/collectives/common.py

@@ -12,6 +12,7 @@
 
 from cudf_polars.dsl.ir import Distinct, GroupBy, Sort
 from cudf_polars.dsl.traversal import traversal
+from cudf_polars.streaming.filter_hint import PushdownFilterHint
 from cudf_polars.streaming.io import StreamingSink
 from cudf_polars.streaming.join import Join
 from cudf_polars.streaming.over import Over
@@ -107,6 +108,7 @@ def __init__(
                 GroupBy,
                 Distinct,
                 Over,
+                PushdownFilterHint,
             )
 
         self.collective_nodes: list[IR] = [

python/cudf_polars/cudf_polars/streaming/actor_graph/core.py

@@ -23,6 +23,7 @@
     generate_ir_sub_network_wrapper,
     metadata_drain_node,
 )
+from cudf_polars.streaming.filter_hint import PushdownFilterHint
 from cudf_polars.streaming.io import StreamingScan
 from cudf_polars.streaming.over import Over
 from cudf_polars.utils.config import SPMDContext
@@ -178,11 +179,12 @@ def _mark_children_unbounded(node: IR) -> None:
     for node in traversal([ir]):
         if node in unbounded:
             _mark_children_unbounded(node)
-        elif isinstance(node, (Union, Join, Over)):
+        elif isinstance(node, (Union, Join, Over, PushdownFilterHint)):
             # Union processes children sequentially; Join may broadcast one
             # side; Over buffers (or samples-then-replays) its input before
-            # producing output. In every case the input source needs
-            # unbounded fanout so other consumers don't block it.
+            # producing output; PushdownFilterHint similarly might buffer
+            # then replay. In every case the input source needs unbounded
+            # fanout so other consumers don't block it.
             _mark_children_unbounded(node)
         elif len(node.children) > 1:
             # Check if this node is doing any broadcasting.