fix(test): support out-of-nanosecond-range timestamps in unit test comparisons

sqlmesh/core/test/definition.py

@@ -263,12 +263,9 @@ def assert_equal(
         for col, value in object_sentinel_values.items():
             try:
                 # can't use `isinstance()` here - https://stackoverflow.com/a/68743663/1707525
-                if type(value) is datetime.date:
-                    expected[col] = pd.to_datetime(expected[col]).dt.date
-                elif type(value) is datetime.time:
-                    expected[col] = pd.to_datetime(expected[col]).dt.time
-                elif type(value) is datetime.datetime:
-                    expected[col] = pd.to_datetime(expected[col]).dt.to_pydatetime()
+                value_type = type(value)
+                if value_type in (datetime.date, datetime.time, datetime.datetime):
+                    expected[col] = _parse_expected_datetime_column(expected[col], value_type)
             except Exception as e:
                 from sqlmesh.core.console import get_console
 
@@ -1014,6 +1011,30 @@ def _raise_error(msg: str, path: Path | None = None) -> None:
     raise TestError(f"Failed to run test:\n{msg}")
 
 
+def _parse_expected_datetime_column(series: pd.Series, target_type: type) -> pd.Series:
+    """Convert a series of expected values to python ``date``/``time``/``datetime``.
+
+    Falls back to microsecond resolution when pandas' default nanosecond
+    parsing overflows. SQL ``TIMESTAMP`` columns can carry values outside
+    pandas' default ``datetime64[ns]`` range (1677-09-21..2262-04-11), so
+    unit tests may compare against values like ``0001-01-01`` which are
+    valid in the database but overflow the default resolution.
+    """
+    import pandas as pd
+    from pandas.errors import OutOfBoundsDatetime
+
+    try:
+        parsed = pd.to_datetime(series)
+    except OutOfBoundsDatetime:
+        parsed = series.astype("datetime64[us]")
+
+    if target_type is datetime.date:
+        return parsed.dt.date
+    if target_type is datetime.time:
+        return parsed.dt.time
+    return parsed.dt.to_pydatetime()
+
+
 def _normalize_df_value(value: t.Any) -> t.Any:
     """Normalize data in a pandas dataframe so ruamel and sqlglot can deal with it."""
     import numpy as np

tests/core/test_test.py

@@ -2931,6 +2931,46 @@ def test_timestamp_normalization() -> None:
     )
 
 
+def test_out_of_bounds_nanosecond_timestamp_comparison(mocker: MockerFixture) -> None:
+    # https://github.com/TobikoData/sqlmesh/issues/5929
+    # Engines like Redshift may return a TIMESTAMP column as an object-dtype
+    # series of python `datetime.datetime` instances. Values outside pandas'
+    # default `datetime64[ns]` range (1677-09-21..2262-04-11) - which SQL
+    # `TIMESTAMP` fully supports - previously raised `OutOfBoundsDatetime`
+    # while parsing the expected values, producing a "Failed to convert
+    # expected value into `datetime`" warning and a false mismatch on values
+    # whose repr survives str-coercion (the values below happen to compare
+    # equal via `str()`, so the mismatch was silent).
+    test = _create_test(
+        body=load_yaml(
+            """
+test_foo:
+  model: sushi.foo
+  outputs:
+    query:
+      - ts_col: "0001-01-01 00:00:00"
+      - ts_col: "9999-12-31 23:59:59"
+            """
+        ),
+        test_name="test_foo",
+        model=_create_model("SELECT ts_col FROM raw"),
+        context=Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))),
+    )
+    actual = pd.DataFrame(
+        {
+            "ts_col": pd.Series(
+                [datetime.datetime(1, 1, 1), datetime.datetime(9999, 12, 31, 23, 59, 59)],
+                dtype=object,
+            )
+        }
+    )
+    expected = pd.DataFrame({"ts_col": ["0001-01-01 00:00:00", "9999-12-31 23:59:59"]})
+    log_warning = mocker.spy(get_console(), "log_warning")
+    test.assert_equal(expected=expected, actual=actual, sort=False)
+    for call_args in log_warning.call_args_list:
+        assert "Failed to convert expected value" not in call_args.args[0]
+
+
 @use_terminal_console
 def test_disable_test_logging_if_no_tests_found(mocker: MockerFixture, tmp_path: Path) -> None:
     init_example_project(tmp_path, engine_type="duckdb")