def udf_entrypoint() -> int:
    """Parallel processing (faster for more CPU-heavy UDFs)."""
    # Load UDF info from stdin
    udf_info: UdfInfo = load(stdin.buffer)

    query = udf_info["query"]
    batching = udf_info["batching"]
    is_generator = udf_info["is_generator"]

    download_cb = get_download_callback()
    processed_cb = get_processed_callback()
    generated_cb = get_generated_callback(is_generator)

    wh_cls, wh_args, wh_kwargs = udf_info["warehouse_clone_params"]
    warehouse: AbstractWarehouse = wh_cls(*wh_args, **wh_kwargs)

    id_col = get_query_id_column(query)

    with contextlib.closing(
        batching(warehouse.dataset_select_paginated, query, id_col=id_col)
    ) as udf_inputs:
        try:
            UDFDispatcher(udf_info).run_udf(
                udf_inputs,
                ids_only=id_col is not None,
                download_cb=download_cb,
                processed_cb=processed_cb,
                generated_cb=generated_cb,
            )
        finally:
            download_cb.close()
            processed_cb.close()
            generated_cb.close()

    return 0
Parallel processing (faster for more CPU-heavy UDFs).
src/datachain/query/dispatch.py