Title: `Dataset.map()` breaks when `function` calls `import polars as pl` and `num_proc`>1: "UnboundLocalError: cannot access local variable 'pl' where it is not associated with a value"
State: closed

### Describe the bug

# Repro
These two conditions seem to consistently reproduce the issue:
- function passed to `Dataset.map()` explicitly or implicitly calls `import polars as pl`
- `num_proc` > 1

# Trace
```
RemoteTraceback                           Traceback (most recent call last)
RemoteTraceback: 
"""
Traceback (most recent call last):
  File "c:\Users\{redacted}\.venv\Lib\site-packages\multiprocess\pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
                    ^^^^^^^^^^^^^^^^^^^
  File "c:\Users\{redacted}\.venv\Lib\site-packages\datasets\utils\py_utils.py", line 586, in _write_generator_to_queue
    for i, result in enumerate(func(**kwargs)):
                     ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\Users\{redacted}\.venv\Lib\site-packages\datasets\arrow_dataset.py", line 3687, in _map_single
    and isinstance(example, pl.DataFrame)
                            ^^
UnboundLocalError: cannot access local variable 'pl' where it is not associated with a value
"""

The above exception was the direct cause of the following exception:

UnboundLocalError                         Traceback (most recent call last)
Cell In[2], [line 9](vscode-notebook-cell:?execution_count=2&line=9)
      6     import polars as pl
      7     return {'squared': sample['n'] ** 2}
----> [9](vscode-notebook-cell:?execution_count=2&line=9) ds.map(square, num_proc=2)

File c:\Users\{redacted}\.venv\Lib\site-packages\datasets\arrow_dataset.py:562, in transmit_format.<locals>.wrapper(*args, **kwargs)
    555 self_format = {
    556     "type": self._format_type,
    557     "format_kwargs": self._format_kwargs,
    558     "columns": self._format_columns,
    559     "output_all_columns": self._output_all_columns,
    560 }
    561 # apply actual function
--> [562](file:///C:/Users/{redacted}/.venv/Lib/site-packages/datasets/arrow_dataset.py:562) out: Union["Dataset", "DatasetDict"] = func(self, *args, **kwargs)
    563 datasets: list["Dataset"] = list(out.values()) if isinstance(out, dict) else [out]
    564 # re-apply format to the output

File c:\Users\{redacted}\.venv\Lib\site-packages\datasets\arrow_dataset.py:3334, in Dataset.map(self, function, with_indices, with_rank, input_columns, batched, batch_size, drop_last_batch, remove_columns, keep_in_memory, load_from_cache_file, cache_file_name, writer_batch_size, features, disable_nullable, fn_kwargs, num_proc, suffix_template, new_fingerprint, desc, try_original_type)
   3331 os.environ = prev_env
   3332 logger.info(f"Spawning {num_proc} processes")
-> [3334](file:///C:/Users/{redacted}/.venv/Lib/site-packages/datasets/arrow_dataset.py:3334) for rank, done, content in iflatmap_unordered(
   3335     pool, Dataset._map_single, kwargs_iterable=unprocessed_kwargs_per_job
   3336 ):
   3337     check_if_shard_done(rank, done, content)
   3339 pool.close()

File c:\Users\{redacted}\.venv\Lib\site-packages\datasets\utils\py_utils.py:626, in iflatmap_unordered(pool, func, kwargs_iterable)
    623 finally:
    624     if not pool_changed:
    625         # we get the result in case there's an error to raise
--> [626](file:///C:/Users/{redacted}/.venv/Lib/site-packages/datasets/utils/py_utils.py:626)         [async_result.get(timeout=0.05) for async_result in async_results]

File c:\Users\{redacted}\.venv\Lib\site-packages\multiprocess\pool.py:774, in ApplyResult.get(self, timeout)
    772     return self._value
    773 else:
--> [774](file:///C:/Users/{redacted}/.venv/Lib/site-packages/multiprocess/pool.py:774)     raise self._value

UnboundLocalError: cannot access local variable 'pl' where it is not associated with a value
```

# Why `import polars` in a worker function?
To my knowledge `Dataset.map()` doesn't support a worker init function, and objects useful inside a function aren't always pickable, so I commonly use this pattern to essentially construct the unpickable object in a worker process:

```python
def func(example, **kwargs):
    if 'my_unpickable_object' not in globals():
        from my_module import MyClass
        my_unpickable_object = MyClass(**kwargs)
    return {'newcol': my_unpickable_object.calculate_something(example['n'])}

ds = Dataset.load_from_disk(...)
ds.map(func, num_proc=2, ...)
```
and here `from my_module import MyClass` may implicitly call `import polars as pl` e.g. when `my_module.py` has that line, or when it imports some other module containing `import polars as pl`.

# A workaround
Things seem to work ok if I don't do any import inside `func`, but instead pass a class already imported outside `func` as a contructor, like below. Although I'm unsure how many scenarios this workaround can cover.

```python
from my_module import MyClass

def func(example, constructor=MyClass, **kwargs):
    if 'my_unpickable_object' not in globals():
        my_unpickable_object = constructor(**kwargs)
    return {'newcol': my_unpickable_object.calculate_something(example['n'])}

ds = Dataset.load_from_disk(...)
ds.map(func, num_proc=2, ...)
```

# My speculations
Before the crash point, on line 3656 of `arrow_dataset.py` it reads:
```python
if config.POLARS_AVAILABLE and "polars" in sys.modules:
    import polars as pl
````
My guess is these conditions are somehow messed up in a worker process, and it ended up not entering the `if` block.

### Steps to reproduce the bug

```python
from datasets import Dataset

ds = Dataset.from_dict({'n': list(range(10_000))})

def square(sample):
    import polars as pl
    return {'squared': sample['n'] ** 2}

ds.map(square, num_proc=2)
```

### Expected behavior

```
Dataset({
    features: ['n', 'squared'],
    num_rows: 10000
})
```

### Environment info

- `datasets` version: 4.5.0
- Platform: Windows-11-10.0.26200-SP0
- Python version: 3.12.10
- `huggingface_hub` version: 0.31.2
- PyArrow version: 21.0.0
- Pandas version: 2.2.3
- `fsspec` version: 2025.3.0

Comments: