Repository: pola-rs/polars (language: python)

Issue:
`offset_by` over Feb 29 creates "duplicate" dates that Polars can't de-dupe.
### Checks

- [X] I have checked that this issue has not already been reported.
- [X] I have confirmed this bug exists on the [latest version](https://pypi.org/project/polars/) of Polars.

### Reproducible example

```python
import polars as pl

# set up some dates over Feb 28 and 29, 2020
df20 = (
	pl.select(
		t=pl.datetime_range(pl.datetime(2020,2,28), pl.datetime(2020,3,1), closed="left", time_unit="ms", interval="8h", time_zone="UTC"),
	).with_columns(
		x=pl.int_range(pl.len())
	)
)
# now use offset_by a year into the past and into the future
df19 = df20.with_columns(pl.col("t").dt.offset_by("-1y"))
df21 = df20.with_columns(pl.col("t").dt.offset_by("1y"))
```

```py
# now let's see what we got...
display(df19)
```
```
# OK, we've got 2 × dates for Feb 28, seems reasonable...
shape: (6, 2)
┌─────────────────────────┬─────┐
│ t                       ┆ x   │
│ ---                     ┆ --- │
│ datetime[ms, UTC]       ┆ i64 │
╞═════════════════════════╪═════╡
│ 2019-02-28 00:00:00 UTC ┆ 0   │
│ 2019-02-28 08:00:00 UTC ┆ 1   │
│ 2019-02-28 16:00:00 UTC ┆ 2   │
│ 2019-02-28 00:00:00 UTC ┆ 3   │
│ 2019-02-28 08:00:00 UTC ┆ 4   │
│ 2019-02-28 16:00:00 UTC ┆ 5   │
└─────────────────────────┴─────┘
```
```py
# ok, let's dedupe...
display(df19.unique("t", keep="first"))
```
```
# Wait what?! according to `unique()`, none of these rows are duplicates!
shape: (6, 2)
┌─────────────────────────┬─────┐
│ t                       ┆ x   │
│ ---                     ┆ --- │
│ datetime[ms, UTC]       ┆ i64 │
╞═════════════════════════╪═════╡
│ 2019-02-28 00:00:00 UTC ┆ 0   │
│ 2019-02-28 08:00:00 UTC ┆ 1   │
│ 2019-02-28 16:00:00 UTC ┆ 2   │
│ 2019-02-28 00:00:00 UTC ┆ 3   │
│ 2019-02-28 08:00:00 UTC ┆ 4   │
│ 2019-02-28 16:00:00 UTC ┆ 5   │
└─────────────────────────┴─────┘
```
The expected behaviour would be for all rows to be reported as duplicates, and my `unique` call to return 3 rows.

df21 exhibits the same incorrect behaviour as df19. df20 behaves fine (6 rows for feb 28/29 for both the above calls)

-----

Polars not picking them up as duplicates seems pervasive, my example here is with `unique` but the same behaviour happens with everything else I've tried.

I noticed something was fishy from finding duplicate rows in the output of `align_frames`, and I think I've narrowed it back to the source probably at `offset_by`?

### Log output

```shell
group_by keys are sorted; running sorted key fast path
group_by keys are sorted; running sorted key fast path
group_by keys are sorted; running sorted key fast path
```


### Issue description

When doing `offset_by("1y")` on dates on Feb 29, the resulting Feb 28 datetimes are not considered deduplicates of "identical" other datetimes on Feb 28. This means the normal approach to deduplicating them silently fails, which is unexpected, to say the least.

### Expected behavior

```py
display(df19.unique("t", keep="first"))
```
```
shape: (~6~ 3, 2)
┌─────────────────────────┬─────┐
│ t                       ┆ x   │
│ ---                     ┆ --- │
│ datetime[ms, UTC]       ┆ i64 │
╞═════════════════════════╪═════╡
│ 2019-02-28 00:00:00 UTC ┆ 0   │
│ 2019-02-28 08:00:00 UTC ┆ 1   │
│ 2019-02-28 16:00:00 UTC ┆ 2   │
└─────────────────────────┴─────┘
```

### Installed versions

<details>

```
--------Version info---------
Polars:              1.12.0
Index type:          UInt32
Platform:            Linux-6.9.3-76060903-generic-x86_64-with-glibc2.35
Python:              3.12.4 (main, Jun  6 2024, 18:26:44) [GCC 11.4.0]
LTS CPU:             False

----Optional dependencies----
adbc_driver_manager  <not installed>
altair               5.4.1
cloudpickle          <not installed>
connectorx           <not installed>
deltalake            <not installed>
fastexcel            <not installed>
fsspec               <not installed>
gevent               <not installed>
great_tables         <not installed>
matplotlib           <not installed>
nest_asyncio         1.6.0
numpy                2.1.2
openpyxl             <not installed>
pandas               2.2.3
pyarrow              17.0.0
pydantic             <not installed>
pyiceberg            <not installed>
sqlalchemy           <not installed>
torch                <not installed>
xlsx2csv             <not installed>
xlsxwriter           <not installed>
```

</details>



Discussion / hints:
thanks for the report, will take a look
looks like it's a sorted flag being kept erroneously
```python

In [6]:  df20.with_columns(pl.col("t").dt.offset_by("-1y"))
Out[6]:
shape: (6, 2)
┌─────────────────────────┬─────┐
│ t                       ┆ x   │
│ ---                     ┆ --- │
│ datetime[ms, UTC]       ┆ i64 │
╞═════════════════════════╪═════╡
│ 2019-02-28 00:00:00 UTC ┆ 0   │
│ 2019-02-28 08:00:00 UTC ┆ 1   │
│ 2019-02-28 16:00:00 UTC ┆ 2   │
│ 2019-02-28 00:00:00 UTC ┆ 3   │
│ 2019-02-28 08:00:00 UTC ┆ 4   │
│ 2019-02-28 16:00:00 UTC ┆ 5   │
└─────────────────────────┴─────┘

In [7]:  df20.with_columns(pl.col("t").dt.offset_by("-1y")).flags
Out[7]:
{'t': {'SORTED_ASC': True, 'SORTED_DESC': False},
 'x': {'SORTED_ASC': True, 'SORTED_DESC': False}}
```

Fix patch:
diff --git a/crates/polars-time/src/offset_by.rs b/crates/polars-time/src/offset_by.rs
index 00485a59d50d..dcdd1e1666ff 100644
--- a/crates/polars-time/src/offset_by.rs
+++ b/crates/polars-time/src/offset_by.rs
@@ -58,20 +58,35 @@ fn apply_offsets_to_datetime(
 }
 
 pub fn impl_offset_by(ts: &Series, offsets: &Series) -> PolarsResult<Series> {
-    let preserve_sortedness: bool;
     let offsets = offsets.str()?;
-    let out = match ts.dtype() {
+    let dtype = ts.dtype();
+
+    // Sortedness may not be preserved for non-constant durations,
+    // see https://github.com/pola-rs/polars/issues/19608 for a counterexample.
+    // Constant durations (e.g. 2 hours) always preserve sortedness.
+    let tz = match dtype {
+        DataType::Date => None,
+        DataType::Datetime(_, tz) => tz.clone(),
+        _ => polars_bail!(InvalidOperation: "expected Date or Datetime, got {}", dtype),
+    };
+    let preserve_sortedness = match offsets.len() {
+        1 => match offsets.get(0) {
+            Some(offset) => {
+                let offset = Duration::parse(offset);
+                offset.is_constant_duration(tz.as_deref())
+            },
+            None => false,
+        },
+        _ => false,
+    };
+
+    let out = match dtype {
         DataType::Date => {
             let ts = ts
                 .cast(&DataType::Datetime(TimeUnit::Milliseconds, None))
                 .unwrap();
             let datetime = ts.datetime().unwrap();
             let out = apply_offsets_to_datetime(datetime, offsets, None)?;
-            // sortedness is only guaranteed to be preserved if a constant offset is being added to every datetime
-            preserve_sortedness = match offsets.len() {
-                1 => offsets.get(0).is_some(),
-                _ => false,
-            };
             out.cast(&DataType::Datetime(TimeUnit::Milliseconds, None))
                 .unwrap()
                 .cast(&DataType::Date)
@@ -86,21 +101,6 @@ pub fn impl_offset_by(ts: &Series, offsets: &Series) -> PolarsResult<Series> {
                 },
                 _ => apply_offsets_to_datetime(datetime, offsets, None)?,
             };
-            // Sortedness may not be preserved when crossing daylight savings time boundaries
-            // for calendar-aware durations.
-            // Constant durations (e.g. 2 hours) always preserve sortedness.
-            preserve_sortedness = match offsets.len() {
-                1 => match offsets.get(0) {
-                    Some(offset) => {
-                        let offset = Duration::parse(offset);
-                        tz.is_none()
-                            || tz.as_deref() == Some("UTC")
-                            || offset.is_constant_duration(tz.as_deref())
-                    },
-                    None => false,
-                },
-                _ => false,
-            };
             out.cast(&DataType::Datetime(*tu, tz.clone()))
         },
         dt => polars_bail!(


Test patch:
diff --git a/py-polars/tests/unit/operations/namespaces/temporal/test_datetime.py b/py-polars/tests/unit/operations/namespaces/temporal/test_datetime.py
index 6bbd10828496..5852d31aed52 100644
--- a/py-polars/tests/unit/operations/namespaces/temporal/test_datetime.py
+++ b/py-polars/tests/unit/operations/namespaces/temporal/test_datetime.py
@@ -178,9 +178,9 @@ def test_local_time_before_epoch(time_unit: TimeUnit) -> None:
         (None, "1d", True),
         ("Europe/London", "1d", False),
         ("UTC", "1d", True),
-        (None, "1mo", True),
-        ("Europe/London", "1mo", False),
-        ("UTC", "1mo", True),
+        (None, "1m", True),
+        ("Europe/London", "1m", True),
+        ("UTC", "1m", True),
         (None, "1w", True),
         ("Europe/London", "1w", False),
         ("UTC", "1w", True),
diff --git a/py-polars/tests/unit/operations/namespaces/temporal/test_offset_by.py b/py-polars/tests/unit/operations/namespaces/temporal/test_offset_by.py
index 79adde1e7b81..478025cf9a83 100644
--- a/py-polars/tests/unit/operations/namespaces/temporal/test_offset_by.py
+++ b/py-polars/tests/unit/operations/namespaces/temporal/test_offset_by.py
@@ -6,7 +6,7 @@
 import pytest
 
 import polars as pl
-from polars.testing import assert_series_equal
+from polars.testing import assert_frame_equal, assert_series_equal
 
 if TYPE_CHECKING:
     from polars._typing import TimeUnit
@@ -116,3 +116,30 @@ def test_datetime_offset_by(
         time_zone
     )
     assert_series_equal(result, expected)
+
+
+def test_offset_by_unique_29_feb_19608() -> None:
+    df20 = pl.select(
+        t=pl.datetime_range(
+            pl.datetime(2020, 2, 28),
+            pl.datetime(2020, 3, 1),
+            closed="left",
+            time_unit="ms",
+            interval="8h",
+            time_zone="UTC",
+        ),
+    ).with_columns(x=pl.int_range(pl.len()))
+    df19 = df20.with_columns(pl.col("t").dt.offset_by("-1y"))
+    result = df19.unique("t", keep="first").sort("t")
+    expected = pl.DataFrame(
+        {
+            "t": [
+                datetime(2019, 2, 28),
+                datetime(2019, 2, 28, 8),
+                datetime(2019, 2, 28, 16),
+            ],
+            "x": [0, 1, 2],
+        },
+        schema_overrides={"t": pl.Datetime("ms", "UTC")},
+    )
+    assert_frame_equal(result, expected)