Reuse existing semi-join domains

docs/cudf/source/cudf_polars/api.md

@@ -67,6 +67,7 @@ Most users interact with them through `StreamingOptions` fields rather than dire
 .. automodule:: cudf_polars.utils.config
    :members:
       DynamicPlanningOptions,
+      JoinDomainPrefilterOptions,
       MemoryResourceConfig,
       ParquetOptions,
       StreamingExecutor,

docs/cudf/source/cudf_polars/options.md

@@ -108,6 +108,7 @@ Environment variables follow these patterns:
 | `broadcast_limit`        | Maximum number of bytes for broadcast joins.                                                                                                        | auto        |
 | `target_partition_size`  | Target partition size in bytes. Used for IO and dynamic planning. `0` means auto.                                                                   | auto        |
 | `dynamic_planning`       | Dynamic planning configuration, dict or {class}`~cudf_polars.utils.config.DynamicPlanningOptions`. `None` disables.                                 | enabled     |
+| `join_domain_prefilter`  | Join-domain prefilter configuration, dict or {class}`~cudf_polars.utils.config.JoinDomainPrefilterOptions`. `None` disables.                        | enabled     |
 | `sink_to_directory`      | Whether `.sink_*()` writes its output as a directory. The `spmd`, `ray`, and `dask` engines always use `True`; passing `False` raises `ValueError`. | `True`      |
 
 ### Category: `engine`

python/cudf_polars/cudf_polars/engine/options.py

@@ -25,6 +25,7 @@
 
     from cudf_polars.utils.config import (
         DynamicPlanningOptions,
+        JoinDomainPrefilterOptions,
         ParquetOptions,
     )
 
@@ -247,6 +248,14 @@ class StreamingOptions:
         Env: ``CUDF_POLARS__EXECUTOR__DYNAMIC_PLANNING``.
         Default: enabled.
         Category: executor.
+    join_domain_prefilter
+        Join-domain prefilter config, dict or
+        :class:`~cudf_polars.utils.config.JoinDomainPrefilterOptions`. ``None``
+        disables the rewrite.
+        Env: ``CUDF_POLARS__EXECUTOR__JOIN_DOMAIN_PREFILTER`` and
+        ``CUDF_POLARS__EXECUTOR__JOIN_DOMAIN_PREFILTER__*``.
+        Default: enabled.
+        Category: executor.
     sink_to_directory
         Whether multi-partition sink operations should write to a directory
         rather than a single file. The ``spmd``/``ray``/``dask`` engines
@@ -341,6 +350,9 @@ class StreamingOptions:
     dynamic_planning: dict[str, Any] | DynamicPlanningOptions | None | Unspecified = (
         _opt("executor")
     )
+    join_domain_prefilter: (
+        dict[str, Any] | JoinDomainPrefilterOptions | None | Unspecified
+    ) = _opt("executor")
     sink_to_directory: bool | Unspecified = _opt(
         "executor", "CUDF_POLARS__EXECUTOR__SINK_TO_DIRECTORY", parse_boolean
     )