Duplicate values from .cat.get_categories

[uemurax]

Checks

• I have checked that this issue has not already been reported.
• I have confirmed this bug exists on the latest version of Polars.

Reproducible example

import polars as pl

df = pl.DataFrame([
    pl.Series('e', ['a', 'b'] * 500_000, pl.Enum(['a', 'b'])),
])
df.write_parquet('data.parquet')
df_1 = pl.read_parquet('data.parquet')
print(df_1.select(pl.col('e').cat.get_categories()))

Log output

shape: (6, 1)
┌─────┐
│ e   │
│ --- │
│ str │
╞═════╡
│ a   │
│ b   │
│ a   │
│ b   │
│ a   │
│ b   │
└─────┘

Issue description

Duplicate values are returned from .cat.get_categories. This seems to happen when a large DataFrame containing Enum columns is saved as a parquet and loaded back.

Expected behavior

shape: (2, 1)
┌─────┐
│ e   │
│ --- │
│ str │
╞═════╡
│ a   │
│ b   │
└─────┘

Installed versions

--------Version info---------
Polars:               0.20.31
Index type:           UInt32
Platform:             Linux-5.15.146.1-microsoft-standard-WSL2-x86_64-with-glibc2.38
Python:               3.11.5 (main, Sep 11 2023, 13:54:46) [GCC 11.2.0]

----Optional dependencies----
adbc_driver_manager:  <not installed>
cloudpickle:          <not installed>
connectorx:           <not installed>
deltalake:            <not installed>
fastexcel:            <not installed>
fsspec:               <not installed>
gevent:               <not installed>
hvplot:               <not installed>
matplotlib:           <not installed>
nest_asyncio:         <not installed>
numpy:                <not installed>
openpyxl:             <not installed>
pandas:               <not installed>
pyarrow:              <not installed>
pydantic:             <not installed>
pyiceberg:            <not installed>
pyxlsb:               <not installed>
sqlalchemy:           <not installed>
torch:                <not installed>
xlsx2csv:             <not installed>
xlsxwriter:           <not installed>

[c-peters]

After some digging, problem occurs here. The big parquet file gets split into multiple row groups, which get read into a dataframe of multiple chunks. The schema is correct of each single one is correct. However, we evaluate the get_categories in parallel over each chunk and them combine the results into one which results in this outcome.

polars/crates/polars-lazy/src/physical_plan/executors/projection.rs

Lines 26 to 44 in 4584b45

	let df = if self.streamable
	&& df.n_chunks() > 1
	&& df.height() > POOL.current_num_threads() * 2
	&& self.options.run_parallel
	{
	let chunks = df.split_chunks().collect::<Vec<_>>();
	let iter = chunks.into_par_iter().map(|mut df| {
	let selected_cols = evaluate_physical_expressions(
	&mut df,
	&self.expr,
	state,
	self.has_windows,
	self.options.run_parallel,
	)?;
	check_expand_literals(selected_cols, df.is_empty(), self.options)
	});
	let df = POOL.install(|| iter.collect::<PolarsResult<Vec<_>>>())?;
	accumulate_dataframes_vertical_unchecked(df)