Configuration#

def set_runner_native(num_threads: int | None = None) -> Runner[PartitionT]:
    """Configure Daft to execute dataframes using native multi-threaded processing.

    This is the default execution mode for Daft.

    Returns:
        Runner[PartitionT]: A runner object with the native runner's configuration.

    Note:
        Can also be configured via environment variable: DAFT_RUNNER=native
    """
    return _set_runner_native(num_threads)

def set_runner_ray(
    address: str | None = None,
    noop_if_initialized: bool = False,
    force_client_mode: bool = False,
    *,
    downscale_enabled: bool | None = None,
    downscale_idle_seconds: int | None = None,
    min_survivor_workers: int | None = None,
    pending_release_exclude_seconds: int | None = None,
    worker_startup_timeout: int | None = None,
) -> Runner[PartitionT]:
    """Configure Daft to execute dataframes using the Ray distributed computing framework.

    Args:
        address: Ray cluster address to connect to. If None, connects to or starts a local Ray instance.
        noop_if_initialized: If True, skip initialization if Ray is already running.
        force_client_mode: If True, forces Ray to run in client mode.
        downscale_enabled: Enable/disable retiring idle Ray workers (scale-in). If not provided,
            falls back to the ``DAFT_AUTOSCALING_DOWNSCALE_ENABLED`` environment variable (default: False).
        downscale_idle_seconds: Minimum number of seconds a worker must be idle before it becomes eligible
            for retirement. If not provided, falls back to ``DAFT_AUTOSCALING_DOWNSCALE_IDLE_SECONDS``
            (default: 60).
        min_survivor_workers: Minimum number of Ray workers to keep alive even if they are idle.
            If not provided, falls back to ``DAFT_AUTOSCALING_MIN_SURVIVOR_WORKERS`` (default: 1).
        pending_release_exclude_seconds: Grace period (TTL) for recently-released worker IDs during
            worker discovery, to prevent the autoscaler from immediately respawning them. If not
            provided, falls back to ``DAFT_AUTOSCALING_PENDING_RELEASE_EXCLUDE_SECONDS`` (default: 120).
        worker_startup_timeout: Timeout in seconds for Ray worker actors to report their addresses during startup.
            Can also be configured via the ``DAFT_RAY_WORKER_STARTUP_TIMEOUT`` environment variable.

    Returns:
        Runner[PartitionT]: A runner object with the Ray runner's configurations.

    Note:
        Can also be configured via environment variable: DAFT_RUNNER=ray
    """
    # Allow programmatic configuration of autoscaling/downscaling behavior via `daft.set_runner_ray`.
    # These settings are still backed by environment variables so they can propagate to the Rust
    # scheduler/worker-manager components without threading configuration throughout the stack.
    if downscale_enabled is not None:
        os.environ["DAFT_AUTOSCALING_DOWNSCALE_ENABLED"] = "1" if downscale_enabled else "0"
    if downscale_idle_seconds is not None:
        if downscale_idle_seconds < 0:
            raise ValueError("downscale_idle_seconds must be >= 0")
        os.environ["DAFT_AUTOSCALING_DOWNSCALE_IDLE_SECONDS"] = str(downscale_idle_seconds)
    if min_survivor_workers is not None:
        if min_survivor_workers < 0:
            raise ValueError("min_survivor_workers must be >= 0")
        os.environ["DAFT_AUTOSCALING_MIN_SURVIVOR_WORKERS"] = str(min_survivor_workers)
    if pending_release_exclude_seconds is not None:
        if pending_release_exclude_seconds < 0:
            raise ValueError("pending_release_exclude_seconds must be >= 0")
        os.environ["DAFT_AUTOSCALING_PENDING_RELEASE_EXCLUDE_SECONDS"] = str(pending_release_exclude_seconds)

    return _set_runner_ray(
        address=address,
        noop_if_initialized=noop_if_initialized,
        force_client_mode=force_client_mode,
        worker_startup_timeout=worker_startup_timeout,
    )

def get_or_create_runner() -> Runner[PartitionT]:
    """Get or create the current runner instance.

    If a runner has already been set, returns it. Otherwise, creates a new
    runner using the default configuration (native) and locks it in.

    Returns:
        Runner[PartitionT]: The current runner instance.

    Note:
        After calling this function, the runner cannot be changed for the
        lifetime of the process. Use ``get_or_infer_runner_type`` to check the
        runner type without this side effect.
    """
    return _get_or_create_runner()

def get_or_infer_runner_type() -> str:
    """Get or infer the runner type.

    This API will get or infer the currently used runner type according to the following strategies:
    1. If the `runner` has been set, return its type directly;
    2. Try to determine whether it's currently running on a ray cluster. If so, consider it to be a ray type;
    3. Try to determine based on `DAFT_RUNNER` env variable.

    Returns:
        str: The runner type ("native" or "ray").
    """
    return _get_or_infer_runner_type()

def set_planning_config(
    config: PyDaftPlanningConfig | None = None,
    default_io_config: IOConfig | None = None,
    enable_strict_filter_pushdown: bool | None = None,
) -> DaftContext:
    """Globally sets various configuration parameters which control Daft plan construction behavior.

    These configuration values are used when a Dataframe is being constructed (e.g. calls to create a Dataframe, or to build on an existing Dataframe).

    Args:
        config: A PyDaftPlanningConfig object to set the config to, before applying other kwargs. Defaults to None which indicates
            that the old (current) config should be used.
        default_io_config: A default IOConfig to use in the absence of one being explicitly passed into any Expression (e.g. `.download()`)
            or Dataframe operation (e.g. `daft.read_parquet()`).
    """
    # Replace values in the DaftPlanningConfig with user-specified overrides
    ctx = get_context()
    with ctx._lock:
        old_daft_planning_config = ctx._ctx._daft_planning_config if config is None else config
        new_daft_planning_config = old_daft_planning_config.with_config_values(
            default_io_config=default_io_config, enable_strict_filter_pushdown=enable_strict_filter_pushdown
        )

        ctx._ctx._daft_planning_config = new_daft_planning_config
        return ctx

@contextlib.contextmanager
def planning_config_ctx(**kwargs: Any) -> Generator[None, None, None]:
    """Context manager that wraps set_planning_config to reset the config to its original setting afternwards."""
    original_config = get_context().daft_planning_config
    try:
        set_planning_config(**kwargs)
        yield
    finally:
        set_planning_config(config=original_config)

def set_execution_config(
    config: PyDaftExecutionConfig | None = None,
    enable_scan_task_split_and_merge: bool | None = None,
    scan_tasks_min_size_bytes: int | None = None,
    scan_tasks_max_size_bytes: int | None = None,
    max_sources_per_scan_task: int | None = None,
    broadcast_join_size_bytes_threshold: int | None = None,
    parquet_split_row_groups_max_files: int | None = None,
    hash_join_partition_size_leniency: float | None = None,
    sample_size_for_sort: int | None = None,
    num_preview_rows: int | None = None,
    parquet_target_filesize: int | None = None,
    parquet_target_row_group_size: int | None = None,
    parquet_inflation_factor: float | None = None,
    csv_target_filesize: int | None = None,
    csv_inflation_factor: float | None = None,
    json_target_filesize: int | None = None,
    json_inflation_factor: float | None = None,
    text_inflation_factor: float | None = None,
    shuffle_aggregation_default_partitions: int | None = None,
    partial_aggregation_threshold: int | None = None,
    high_cardinality_aggregation_threshold: float | None = None,
    read_sql_partition_size_bytes: int | None = None,
    default_morsel_size: int | None = None,
    shuffle_algorithm: str | None = None,
    pre_shuffle_merge_threshold: int | None = None,
    pre_shuffle_merge_partition_threshold: int | None = None,
    scantask_max_parallel: int | None = None,
    native_parquet_writer: bool | None = None,
    min_cpu_per_task: float | None = None,
    actor_udf_ready_timeout: int | None = None,
    maintain_order: bool | None = None,
    enable_dynamic_batching: bool | None = None,
    dynamic_batching_strategy: str | None = None,
    flight_shuffle_dirs: list[str] | None = None,
    flight_shuffle_compression: str | None = None,
    enable_multi_glob_path_tasks: bool | None = None,
) -> DaftContext:
    """Globally sets various configuration parameters which control various aspects of Daft execution.

    These configuration values
    are used when a Dataframe is executed (e.g. calls to `DataFrame.write_*`, [DataFrame.collect()](https://docs.daft.ai/en/stable/api/dataframe/#daft.DataFrame.collect) or [DataFrame.show()](https://docs.daft.ai/en/stable/api/dataframe/#daft.DataFrame.select)).

    Args:
        config: A PyDaftExecutionConfig object to set the config to, before applying other kwargs. Defaults to None which indicates
            that the old (current) config should be used.
        enable_scan_task_split_and_merge: Whether to enable scan task split and merge. Defaults to False.
        scan_tasks_min_size_bytes: Minimum size of scan tasks in bytes. Defaults to 96MB.
        scan_tasks_max_size_bytes: Maximum size of scan tasks in bytes. Defaults to 384MB.
        max_sources_per_scan_task: Maximum number of sources per scan task. Defaults to 10.
        parquet_split_row_groups_max_files: Maximum number of files to read in which the row group splitting should happen. (Defaults to 10)
        broadcast_join_size_bytes_threshold: If one side of a join is smaller than this threshold, a broadcast join will be used.
            Default is 10 MiB.
        hash_join_partition_size_leniency: If the left side of a hash join is already correctly partitioned and the right side isn't,
            and the ratio between the left and right size is at least this value, then the right side is repartitioned to have an equal
            number of partitions as the left. Defaults to 0.5.
        sample_size_for_sort: number of elements to sample from each partition when running sort,
            Default is 20.
        num_preview_rows: number of rows to when showing a dataframe preview,
            Default is 8.
        parquet_target_filesize: Target File Size when writing out Parquet Files. Defaults to 512MB
        parquet_target_row_group_size: Target Row Group Size when writing out Parquet Files. Defaults to 128MB
        parquet_inflation_factor: Inflation Factor of parquet files (In-Memory-Size / File-Size) ratio. Defaults to 3.0
        csv_target_filesize: Target File Size when writing out CSV Files. Defaults to 512MB
        csv_inflation_factor: Inflation Factor of CSV files (In-Memory-Size / File-Size) ratio. Defaults to 0.5
        json_target_filesize: Target File Size when writing out JSON Files. Defaults to 512MB
        json_inflation_factor: Inflation Factor of JSON files (In-Memory-Size / File-Size) ratio. Defaults to 0.25
        text_inflation_factor: Inflation Factor of Text files (In-Memory-Size / File-Size) ratio. Defaults to 1.0
        shuffle_aggregation_default_partitions: Maximum number of partitions to create when performing aggregations on the Ray Runner. Defaults to 200, unless the number of input partitions is less than 200.
        partial_aggregation_threshold: Threshold for performing partial aggregations on the Native Runner. Defaults to 10000 rows.
        high_cardinality_aggregation_threshold: Threshold selectivity for performing high cardinality aggregations on the Native Runner. Defaults to 0.8.
        read_sql_partition_size_bytes: Target size of partition when reading from SQL databases. Defaults to 512MB
        default_morsel_size: Default size of morsels used for the new local executor. Defaults to 131072 rows.
        shuffle_algorithm: The shuffle algorithm to use. Defaults to "auto", which will let Daft determine the algorithm. Options are "map_reduce", "pre_shuffle_merge", and "flight_shuffle".
        pre_shuffle_merge_threshold: Memory threshold in bytes for pre-shuffle merge. Defaults to 1GB
        pre_shuffle_merge_partition_threshold: Number of partitions threshold to enable pre-shuffle merge when shuffle_algorithm is "auto". Defaults to 200.
        scantask_max_parallel: Set the max parallelism for running scan tasks simultaneously. Currently, this only works for Native Runner. If set to 0, all available CPUs will be used. Defaults to 8.
        native_parquet_writer: Whether to use the native parquet writer vs the pyarrow parquet writer. Defaults to `True`.
        min_cpu_per_task: Deprecated. This was used by the old Ray runner and has no effect on
            distributed scheduling. It will be removed in v0.8.0.
        actor_udf_ready_timeout: Timeout for UDF actors to be ready. Defaults to 120 seconds.
        maintain_order: Whether to maintain order during execution. Defaults to True. Some blocking sink operators (e.g. write_parquet) won't respect this flag and will always keep maintain_order as false, and propagate to child operators. It's useful to set this to False for running df.collect() when no ordering is required.
        enable_dynamic_batching: Whether to enable dynamic batching. Defaults to False.
        dynamic_batching_strategy: The strategy to use for dynamic batching. Defaults to 'auto'.
        flight_shuffle_dirs: Directories to use for flight shuffle. Defaults to ["/tmp"]. Must not be empty.
        flight_shuffle_compression: Arrow IPC compression for flight shuffle spill files. One of "lz4", "zstd", or "none". Defaults to "lz4". Pass "none" to disable compression; passing Python None leaves the current config unchanged.
        enable_multi_glob_path_tasks: Whether to create multiple glob path tasks in Ray Runner to achieve parallel glob. Defaults to False.
    """
    # Replace values in the DaftExecutionConfig with user-specified overrides
    if min_cpu_per_task is not None:
        warnings.warn(
            "`min_cpu_per_task` is deprecated as of v0.7.0 and has no effect on distributed scheduling. "
            "It will be removed from v0.8.0 onwards.",
            DeprecationWarning,
            stacklevel=2,
        )

    ctx = get_context()
    with ctx._lock:
        old_daft_execution_config = ctx._ctx._daft_execution_config if config is None else config

        new_daft_execution_config = old_daft_execution_config.with_config_values(
            enable_scan_task_split_and_merge=enable_scan_task_split_and_merge,
            scan_tasks_min_size_bytes=scan_tasks_min_size_bytes,
            scan_tasks_max_size_bytes=scan_tasks_max_size_bytes,
            max_sources_per_scan_task=max_sources_per_scan_task,
            parquet_split_row_groups_max_files=parquet_split_row_groups_max_files,
            broadcast_join_size_bytes_threshold=broadcast_join_size_bytes_threshold,
            hash_join_partition_size_leniency=hash_join_partition_size_leniency,
            sample_size_for_sort=sample_size_for_sort,
            num_preview_rows=num_preview_rows,
            parquet_target_filesize=parquet_target_filesize,
            parquet_target_row_group_size=parquet_target_row_group_size,
            parquet_inflation_factor=parquet_inflation_factor,
            csv_target_filesize=csv_target_filesize,
            csv_inflation_factor=csv_inflation_factor,
            json_target_filesize=json_target_filesize,
            json_inflation_factor=json_inflation_factor,
            text_inflation_factor=text_inflation_factor,
            shuffle_aggregation_default_partitions=shuffle_aggregation_default_partitions,
            partial_aggregation_threshold=partial_aggregation_threshold,
            high_cardinality_aggregation_threshold=high_cardinality_aggregation_threshold,
            read_sql_partition_size_bytes=read_sql_partition_size_bytes,
            default_morsel_size=default_morsel_size,
            shuffle_algorithm=shuffle_algorithm,
            pre_shuffle_merge_threshold=pre_shuffle_merge_threshold,
            pre_shuffle_merge_partition_threshold=pre_shuffle_merge_partition_threshold,
            scantask_max_parallel=scantask_max_parallel,
            native_parquet_writer=native_parquet_writer,
            min_cpu_per_task=min_cpu_per_task,
            actor_udf_ready_timeout=actor_udf_ready_timeout,
            maintain_order=maintain_order,
            enable_dynamic_batching=enable_dynamic_batching,
            dynamic_batching_strategy=dynamic_batching_strategy,
            flight_shuffle_dirs=flight_shuffle_dirs,
            flight_shuffle_compression=flight_shuffle_compression,
            enable_multi_glob_path_tasks=enable_multi_glob_path_tasks,
        )

        ctx._ctx._daft_execution_config = new_daft_execution_config
        return ctx

@contextlib.contextmanager
def execution_config_ctx(**kwargs: Any) -> Generator[None, None, None]:
    """Context manager that wraps set_execution_config to reset the config to its original setting afternwards."""
    original_config = get_context()._ctx._daft_execution_config
    try:
        set_execution_config(**kwargs)
        yield
    finally:
        set_execution_config(config=original_config)