import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({
    "x": range(15),
    "y": [pd.NA] * 10 + [1.0] * 5,
    "group": ["a"] * 5 + ["b"] * 5 + ["c"] * 5
})

ddf = dd.from_pandas(df, npartitions=2).astype({
    "y": "double[pyarrow]"
})
print(ddf.compute())
#      x     y group
# 0    0  <NA>     a
# 1    1  <NA>     a
# 2    2  <NA>     a
# 3    3  <NA>     a
# 4    4  <NA>     a
# 5    5  <NA>     b
# 6    6  <NA>     b
# 7    7  <NA>     b
# 8    8  <NA>     b
# 9    9  <NA>     b
# 10  10   1.0     c
# 11  11   1.0     c
# 12  12   1.0     c
# 13  13   1.0     c
# 14  14   1.0     c

ddf = ddf.groupby("group").agg(x=("x", "mean"), y=("y", "mean"))
print(ddf.compute())
#           x    y
# group           
# a       2.0  NaN
# b       7.0  NaN
# c      12.0  1.0
# (Note that groups with <NA> resulted in NaN rather than <NA>)

# These NaNs are then not recognized as missing values
print(ddf["y"].isnull().sum().compute())
# 0
# (While the expected result is 2).

# This breaks other functions, example further aggregations in Pandas
print(ddf["y"].compute().mean(skipna=True))
# nan
# (While the exected result is 1.0).

# And also further aggreations in Dask, albeit with a different unexpected result
print(ddf["y"].mean(skipna=True).compute())
# 0.0
# (While the exected result is 1.0)