test(query-context): update normalize_df mock signature for new offset param

normalize_df now accepts an optional third sql_shifted_temporal_labels
parameter (passed positionally from processing_time_offsets). Three
pre-existing mocks in this file stubbed normalize_df with a fixed two-argument
lambda, which broke when the call site started passing a third positional
argument.

superset/connectors/sqla/models.py

@@ -99,6 +99,7 @@ from superset.models.helpers import (
     AuditMixinNullable,
     CertificationMixin,
     ExploreMixin,
+    get_effective_hours_offset,
     ImportExportMixin,
     QueryResult,
     SoftDeleteMixin,
@@ -1241,6 +1242,8 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
         time_grain: str | None,
         label: str | None = None,
         template_processor: BaseTemplateProcessor | None = None,
+        apply_dataset_offset: bool = False,
+        sql_shifted_temporal_labels: set[str] | None = None,
     ) -> TimestampExpression | Label:
         """
         Return a SQLAlchemy Core element representation of self to be used in a query.
@@ -1248,6 +1251,8 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
         :param time_grain: Optional time grain, e.g. P1Y
         :param label: alias/label that column is expected to have
         :param template_processor: template processor
+        :param apply_dataset_offset: shift the selected axis before truncation
+        :param sql_shifted_temporal_labels: labels shifted before truncation
         :return: A TimeExpression object wrapped in a Label if supported by db
         """
         label = label or utils.DTTM_ALIAS
@@ -1285,6 +1290,27 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
             col = literal_column(expression, type_=type_)
         else:
             col = column(self.column_name, type_=type_)
+        if (
+            apply_dataset_offset
+            and time_grain
+            and self.table
+            and self.db_engine_spec.supports_temporal_column_shift
+            and (offset_hours := self.table.offset or 0)
+            and not self.table.get_dataset_timezone()
+        ):
+            effective_offset_hours = get_effective_hours_offset(
+                self.db_engine_spec,
+                self.type,
+                offset_hours,
+                db_extra=self.db_extra,
+            )
+            if effective_offset_hours:
+                col = self.db_engine_spec.get_temporal_column_shift_expr(
+                    col,
+                    effective_offset_hours,
+                )
+            if sql_shifted_temporal_labels is not None:
+                sql_shifted_temporal_labels.add(label)
         time_expr = self.db_engine_spec.get_timestamp_expr(col, pdf, time_grain)
         return self.database.make_sqla_column_compatible(time_expr, label)
 
@@ -1954,11 +1980,26 @@ class SqlaTable(
                 )
             ) from ex
 
+    def _shift_temporal_column_if_needed(
+        self,
+        sqla_column: ColumnClause,
+        effective_offset_hours: int,
+    ) -> ColumnClause:
+        """Apply a nonzero effective dataset offset to a temporal expression."""
+        if not effective_offset_hours:
+            return sqla_column
+        return self.db_engine_spec.get_temporal_column_shift_expr(
+            sqla_column,
+            effective_offset_hours,
+        )
+
     def adhoc_column_to_sqla(  # pylint: disable=too-many-locals
         self,
         col: AdhocColumn,
         force_type_check: bool = False,
         template_processor: BaseTemplateProcessor | None = None,
+        apply_dataset_offset: bool = False,
+        sql_shifted_temporal_labels: set[str] | None = None,
     ) -> tuple[ColumnElement, utils.GenericDataType | None]:
         """
         Turn an adhoc column into a sqlalchemy column.
@@ -1968,6 +2009,8 @@ class SqlaTable(
                This is needed to validate if a filter with an adhoc column
                is applicable.
         :param template_processor: template_processor instance
+        :param apply_dataset_offset: shift the selected axis before truncation
+        :param sql_shifted_temporal_labels: labels shifted before truncation
         :returns: A tuple of (SQLAlchemy column, generic column type). The
             generic type is populated when the column type is resolved
             (either because the adhoc column matches a physical column, or
@@ -1984,6 +2027,7 @@ class SqlaTable(
         pdf = None
         is_column_reference = col.get("isColumnReference", False)
         generic_type: utils.GenericDataType | None = None
+        native_type: str | None = None
 
         metadata_lookup_key = self._render_adhoc_expression_for_metadata_lookup(
             sql_expression, template_processor
@@ -1998,6 +2042,7 @@ class SqlaTable(
             is_dttm = col_in_metadata.is_temporal
             pdf = col_in_metadata.python_date_format
             generic_type = col_in_metadata.type_generic
+            native_type = col_in_metadata.type
         else:
             # Column doesn't exist in metadata or is not a reference - treat as ad-hoc
             # expression Note: If isColumnReference=true but column not found, we still
@@ -2060,8 +2105,28 @@ class SqlaTable(
                 # stay unquoted for numeric adhoc expressions like
                 # CAST(... AS BIGINT)).
                 generic_type = col_desc[0].get("type_generic")
+                probed_type = col_desc[0].get("type")
+                native_type = str(probed_type) if probed_type is not None else None
 
         if is_dttm and has_timegrain:
+            if (
+                apply_dataset_offset
+                and self.db_engine_spec.supports_temporal_column_shift
+                and (offset_hours := self.offset or 0)
+                and not self.get_dataset_timezone()
+            ):
+                effective_offset_hours = get_effective_hours_offset(
+                    self.db_engine_spec,
+                    native_type,
+                    offset_hours,
+                    db_extra=self.db_extra,
+                )
+                sqla_column = self._shift_temporal_column_if_needed(
+                    sqla_column,
+                    effective_offset_hours,
+                )
+                if sql_shifted_temporal_labels is not None:
+                    sql_shifted_temporal_labels.add(label)
             sqla_column = self.db_engine_spec.get_timestamp_expr(
                 col=sqla_column,
                 pdf=pdf,

superset/db_engine_specs/base.py

@@ -528,6 +528,7 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
     time_groupby_inline = False
     limit_method = LimitMethod.FORCE_LIMIT
     supports_multivalues_insert = False
+    supports_temporal_column_shift: bool = False
     allows_joins = True
     allows_subqueries = True
     allows_alias_in_select = True
@@ -1204,6 +1205,19 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
 
         return TimestampExpression(time_expr, col, type_=col.type)
 
+    @classmethod
+    def get_temporal_column_shift_expr(
+        cls,
+        col: ColumnClause,
+        offset_hours: int,
+    ) -> TimestampExpression:
+        """Shift a temporal SQL expression by a bounded number of hours."""
+        return TimestampExpression(
+            f"{{col}} + INTERVAL '{offset_hours}' HOUR",
+            col,
+            type_=col.type,
+        )
+
     @classmethod
     def _apply_year_to_dttm(cls, time_expr: str) -> str:
         """

superset/db_engine_specs/postgres.py

@@ -307,6 +307,7 @@ class PostgresEngineSpec(BasicParametersMixin, PostgresBaseEngineSpec):
     supports_catalog = True
     supports_dynamic_catalog = True
     supports_grouping_sets = True
+    supports_temporal_column_shift = True
 
     default_driver = "psycopg2"
     sqlalchemy_uri_placeholder = (

superset/db_engine_specs/sqlite.py

@@ -25,9 +25,14 @@ from typing import Any, TYPE_CHECKING
 from flask_babel import gettext as __
 from sqlalchemy import types
 from sqlalchemy.engine.reflection import Inspector
+from sqlalchemy.sql.elements import ColumnClause
 
 from superset.constants import TimeGrain
-from superset.db_engine_specs.base import BaseEngineSpec, DatabaseCategory
+from superset.db_engine_specs.base import (
+    BaseEngineSpec,
+    DatabaseCategory,
+    TimestampExpression,
+)
 from superset.errors import SupersetErrorType
 
 if TYPE_CHECKING:
@@ -43,6 +48,7 @@ class SqliteEngineSpec(BaseEngineSpec):
 
     disable_ssh_tunneling = True
     supports_multivalues_insert = True
+    supports_temporal_column_shift = True
 
     metadata = {
         "description": "SQLite is a self-contained, serverless SQL database engine.",
@@ -140,6 +146,20 @@ class SqliteEngineSpec(BaseEngineSpec):
             "ELSE printf('%04d-01-01', CAST({col} AS INTEGER)) END)"
         )
 
+    @classmethod
+    def get_temporal_column_shift_expr(
+        cls,
+        col: ColumnClause,
+        offset_hours: int,
+    ) -> TimestampExpression:
+        """Shift a temporal expression with SQLite's datetime modifier syntax."""
+        modifier = f"{offset_hours:+d} hours"
+        return TimestampExpression(
+            f"DATETIME({{col}}, '{modifier}')",
+            col,
+            type_=col.type,
+        )
+
     @classmethod
     def convert_dttm(
         cls, target_type: str, dttm: datetime, db_extra: dict[str, Any] | None = None

superset/models/helpers.py

@@ -177,6 +177,21 @@ SERIES_LIMIT_SUBQ_ALIAS = "series_limit"
 # Offset join column suffix used for joining offset results
 OFFSET_JOIN_COLUMN_SUFFIX = "__offset_join_column_"
 
+
+def get_effective_hours_offset(
+    db_engine_spec: type["BaseEngineSpec"],
+    column_type: str | None,
+    offset_hours: int,
+    db_extra: dict[str, Any] | None = None,
+) -> int:
+    """Return the dataset offset representable by a temporal column's type."""
+    sqla_type = db_engine_spec.get_sqla_column_type(column_type, db_extra=db_extra)
+    if isinstance(sqla_type, sa.Date):
+        # int() deliberately truncates toward zero; // would turn -1h into -24h.
+        return int(offset_hours / 24) * 24
+    return offset_hours
+
+
 # Right suffix used for joining offset results
 R_SUFFIX = "__right_suffix"
 
@@ -1294,6 +1309,7 @@ class QueryResult:  # pylint: disable=too-few-public-methods
         errors: Optional[list[dict[str, Any]]] = None,
         from_dttm: Optional[datetime] = None,
         to_dttm: Optional[datetime] = None,
+        sql_shifted_temporal_labels: set[str] | None = None,
     ) -> None:
         self.df = df
         self.query = query
@@ -1306,6 +1322,7 @@ class QueryResult:  # pylint: disable=too-few-public-methods
         self.errors = errors or []
         self.from_dttm = from_dttm
         self.to_dttm = to_dttm
+        self.sql_shifted_temporal_labels = sql_shifted_temporal_labels or set()
         self.sql_rowcount = len(self.df.index) if not self.df.empty else 0
 
 
@@ -1405,6 +1422,7 @@ class QueryStringExtended(NamedTuple):
     labels_expected: list[str]
     prequeries: list[str]
     sql: str
+    sql_shifted_temporal_labels: set[str]
 
 
 class SqlaQuery(NamedTuple):
@@ -1416,6 +1434,7 @@ class SqlaQuery(NamedTuple):
     labels_expected: list[str]
     prequeries: list[str]
     sqla_query: Select
+    sql_shifted_temporal_labels: set[str]
 
 
 class ExploreMixin:  # pylint: disable=too-many-public-methods
@@ -1770,6 +1789,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             labels_expected=sqlaq.labels_expected,
             prequeries=sqlaq.prequeries,
             sql=sql,
+            sql_shifted_temporal_labels=sqlaq.sql_shifted_temporal_labels,
         )
 
     def _normalize_prequery_result_type(
@@ -1974,6 +1994,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             query=sql,
             errors=errors,
             error_message=error_message,
+            sql_shifted_temporal_labels=query_str_ext.sql_shifted_temporal_labels,
         )
 
     def exc_query(self, qry: Any) -> QueryResult:
@@ -2043,6 +2064,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         df: pd.DataFrame,
         query_object: QueryObject,
         already_collected: set[str],
+        sql_shifted_temporal_labels: set[str] | None = None,
     ) -> list[DateColumn]:
         """``DateColumn`` entries that only need the dataset HOURS OFFSET (and any
         time shift) applied, for temporal columns the database already returns as
@@ -2062,6 +2084,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         ):
             return []
 
+        sql_shifted_temporal_labels = sql_shifted_temporal_labels or set()
         extra: list[DateColumn] = []
         for label in df.columns:
             if label in already_collected or label == DTTM_ALIAS:
@@ -2080,7 +2103,9 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                 extra.append(
                     DateColumn(
                         timestamp_format=None,
-                        offset=self.offset,
+                        offset=(
+                            0 if label in sql_shifted_temporal_labels else self.offset
+                        ),
                         time_shift=query_object.time_shift,
                         col_label=label,
                     )
@@ -2088,15 +2113,22 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                 already_collected.add(label)
         return extra
 
-    def normalize_df(self, df: pd.DataFrame, query_object: QueryObject) -> pd.DataFrame:
+    def normalize_df(
+        self,
+        df: pd.DataFrame,
+        query_object: QueryObject,
+        sql_shifted_temporal_labels: set[str] | None = None,
+    ) -> pd.DataFrame:
         """
         Normalize the dataframe by converting datetime columns and ensuring
         numerical metrics.
 
         :param df: The dataframe to normalize
         :param query_object: The query object with metadata about columns
+        :param sql_shifted_temporal_labels: labels already shifted in generated SQL
         :return: Normalized dataframe
         """
+        sql_shifted_temporal_labels = sql_shifted_temporal_labels or set()
         labels = self._collect_dttm_labels(query_object)
 
         # ``get_dataset_timezone`` lives on ``ExploreMixin``; datasource doubles
@@ -2108,7 +2140,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         dttm_cols = [
             DateColumn(
                 timestamp_format=fmt,
-                offset=self.offset,
+                offset=0 if label in sql_shifted_temporal_labels else self.offset,
                 time_shift=query_object.time_shift,
                 timezone=dataset_timezone,
                 col_label=label,
@@ -2120,7 +2152,9 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             dttm_cols.append(
                 DateColumn.get_legacy_time_column(
                     timestamp_format=self._python_date_format(query_object.granularity),
-                    offset=self.offset,
+                    offset=(
+                        0 if DTTM_ALIAS in sql_shifted_temporal_labels else self.offset
+                    ),
                     time_shift=query_object.time_shift,
                     timezone=dataset_timezone,
                 )
@@ -2128,7 +2162,10 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
 
         dttm_cols.extend(
             self._offset_only_dttm_cols(
-                df, query_object, {col.col_label for col in dttm_cols}
+                df,
+                query_object,
+                {col.col_label for col in dttm_cols},
+                sql_shifted_temporal_labels,
             )
         )
 
@@ -2174,7 +2211,11 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         df = result.df
         if not df.empty:
             # Normalize datetime columns and metrics
-            df = self.normalize_df(df, query_object)
+            df = self.normalize_df(
+                df,
+                query_object,
+                result.sql_shifted_temporal_labels,
+            )
 
             # Process time offsets if requested
             if query_object.time_offsets:
@@ -2484,7 +2525,9 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             else:
                 # 1. normalize df, set dttm column
                 offset_metrics_df = self.normalize_df(
-                    offset_metrics_df, query_object_clone
+                    offset_metrics_df,
+                    query_object_clone,
+                    result.sql_shifted_temporal_labels,
                 )
 
                 # 2. rename extra query columns
@@ -3417,6 +3460,8 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         col: AdhocColumn,
         force_type_check: bool = False,
         template_processor: Optional[BaseTemplateProcessor] = None,
+        apply_dataset_offset: bool = False,
+        sql_shifted_temporal_labels: set[str] | None = None,
     ) -> tuple[ColumnElement, Optional[GenericDataType]]:
         raise NotImplementedError()
 
@@ -3601,6 +3646,12 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                 dataset_timezone = None
 
         if not dataset_timezone and (offset_hours := getattr(self, "offset", 0) or 0):
+            offset_hours = get_effective_hours_offset(
+                self.db_engine_spec,
+                time_col.type,
+                offset_hours,
+                db_extra=self.db_extra,
+            )
             if start_dttm is not None:
                 start_dttm = start_dttm - timedelta(hours=offset_hours)
             if end_dttm is not None:
@@ -3957,6 +4008,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         template_kwargs["applied_filters"] = applied_template_filters
         template_processor = self.get_template_processor(**template_kwargs)
         prequeries: list[str] = []
+        sql_shifted_temporal_labels: set[str] = set()
         orderby = orderby or []
         need_groupby = bool(metrics is not None or groupby)
         metrics = metrics or []
@@ -4065,6 +4117,8 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                 col, _unused = self.adhoc_column_to_sqla(
                     col=adhoc_columns_by_label[col],
                     template_processor=template_processor,
+                    apply_dataset_offset=True,
+                    sql_shifted_temporal_labels=sql_shifted_temporal_labels,
                 )
             elif col in metrics_by_name:
                 col = metrics_by_name[col].get_sqla_col(
@@ -4104,6 +4158,8 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                             time_grain=time_grain,
                             label=selected,
                             template_processor=template_processor,
+                            apply_dataset_offset=True,
+                            sql_shifted_temporal_labels=sql_shifted_temporal_labels,
                         )
                     # if groupby field equals a selected column
                     elif selected in columns_by_name:
@@ -4125,6 +4181,8 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                     outer, _unused = self.adhoc_column_to_sqla(
                         col=selected,
                         template_processor=template_processor,
+                        apply_dataset_offset=True,
+                        sql_shifted_temporal_labels=sql_shifted_temporal_labels,
                     )
                 groupby_all_columns[outer.name] = outer
                 if (
@@ -4165,6 +4223,8 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                     outer, _unused = self.adhoc_column_to_sqla(
                         col=selected,
                         template_processor=template_processor,
+                        apply_dataset_offset=True,
+                        sql_shifted_temporal_labels=sql_shifted_temporal_labels,
                     )
                     select_exprs.append(outer)
                     continue
@@ -4200,7 +4260,10 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
 
             if is_timeseries:
                 timestamp = dttm_col.get_timestamp_expression(
-                    time_grain=time_grain, template_processor=template_processor
+                    time_grain=time_grain,
+                    template_processor=template_processor,
+                    apply_dataset_offset=True,
+                    sql_shifted_temporal_labels=sql_shifted_temporal_labels,
                 )
                 # always put timestamp as the first column
                 select_exprs.insert(0, timestamp)
@@ -4865,4 +4928,5 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             labels_expected=labels_expected,
             sqla_query=qry,
             prequeries=prequeries,
+            sql_shifted_temporal_labels=sql_shifted_temporal_labels,
         )

superset/models/sql_lab.py

@@ -437,6 +437,8 @@ class Query(
         col: "AdhocColumn",  # type: ignore  # noqa: F821
         force_type_check: bool = False,
         template_processor: Optional[BaseTemplateProcessor] = None,
+        apply_dataset_offset: bool = False,
+        sql_shifted_temporal_labels: set[str] | None = None,
     ) -> tuple[ColumnElement, Optional[GenericDataType]]:
         """
         Turn an adhoc column into a sqlalchemy column.

tests/unit_tests/common/test_query_context_processor.py

@@ -1142,7 +1142,7 @@ def test_processing_time_offsets_quarter_offset_shifts_query_window(
 
     datasource.query = fake_query
     datasource.normalize_df = MagicMock(
-        side_effect=lambda offset_df, _query_object: offset_df
+        side_effect=lambda offset_df, _query_object, _labels=None: offset_df
     )
 
     with (
@@ -1236,7 +1236,7 @@ def test_processing_time_offsets_accepts_zero_shift_offset(
 
     datasource.query = fake_query
     datasource.normalize_df = MagicMock(
-        side_effect=lambda offset_df, _query_object: offset_df
+        side_effect=lambda offset_df, _query_object, _labels=None: offset_df
     )
 
     with (
@@ -2344,7 +2344,7 @@ def test_relative_offset_preserves_inner_bounds(
 
     datasource.query = fake_query
     datasource.normalize_df = MagicMock(
-        side_effect=lambda offset_df, _query_object: offset_df
+        side_effect=lambda offset_df, _query_object, _labels=None: offset_df
     )
 
     with (

tests/unit_tests/models/test_hours_offset_bound_truncation.py

@@ -0,0 +1,702 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Regression guards for dataset "Hours offset" bound and grain handling.
+
+Two independent defects are covered here.
+
+Defect 1 -- DATE-column filter bounds use a whole-day effective offset so rendering
+    date-only literals cannot discard a sub-day remainder and move the window.
+
+Defect 2 -- grained axis expressions apply the dataset offset in SQL before time
+    grain truncation. Dataframe normalization suppresses its legacy post-query
+    offset only for labels that were shifted in SQL.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import date, datetime
+
+import pandas as pd
+import pytest
+from flask import Flask
+from pytest_mock import MockerFixture
+from sqlalchemy import column, create_engine, DateTime
+from sqlalchemy.dialects import postgresql, sqlite
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm.session import Session
+from sqlalchemy.pool import StaticPool
+
+from superset.common.query_object import QueryObject
+from superset.connectors.sqla.models import SqlaTable, TableColumn
+from superset.db_engine_specs.base import BaseEngineSpec
+from superset.db_engine_specs.postgres import PostgresEngineSpec
+from superset.db_engine_specs.sqlite import SqliteEngineSpec
+from superset.models.core import Database
+from superset.superset_typing import AdhocColumn, QueryObjectDict
+
+# ---------------------------------------------------------------------------
+# Defect 1 -- DATE-column filter bound literal truncated to day precision
+# ---------------------------------------------------------------------------
+
+
+def _pg_dataset(offset: int, col_type: str) -> SqlaTable:
+    """A Postgres-backed dataset with a single temporal column of ``col_type``
+    and the given dataset Hours ``offset``."""
+    database = Database(
+        id=1,
+        database_name="pg",
+        # A postgres:// URI selects PostgresEngineSpec; the SQL is only compiled,
+        # never executed, so no live server is required.
+        sqlalchemy_uri="postgresql://u:p@localhost:5432/db",
+    )
+    columns = [
+        TableColumn(column_name="loan_date", is_dttm=1, type=col_type),
+        TableColumn(column_name="value", type="INTEGER"),
+    ]
+    return SqlaTable(
+        table_name="loans",
+        columns=columns,
+        main_dttm_col="loan_date",
+        database=database,
+        offset=offset,
+    )
+
+
+def _generated_sql(dataset: SqlaTable, mocker: MockerFixture, app: Flask) -> str:
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
+        return_value=[],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.is_guest_user",
+        return_value=False,
+    )
+    # Requested window: the whole month of August 2026, i.e. [2026-08-01, 2026-09-01).
+    query_obj: QueryObjectDict = {
+        "granularity": "loan_date",
+        "from_dttm": datetime(2026, 8, 1),
+        "to_dttm": datetime(2026, 9, 1),
+        "is_timeseries": False,
+        "filter": [
+            {
+                "col": "loan_date",
+                "op": "TEMPORAL_RANGE",
+                "val": "2026-08-01 : 2026-09-01",
+            }
+        ],
+        "metrics": [],
+        "columns": ["value"],
+    }
+    with app.test_request_context():
+        return dataset.get_query_str_extended(query_obj, mutate=False).sql
+
+
+def test_date_column_hours_offset_does_not_shift_selected_day_window(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """A pure ``DATE`` column stores calendar dates at midnight, so a +1h Hours
+    offset can never move a value across a day boundary: the selected window must
+    stay 2026-08-01 .. 2026-08-31 (identical to offset 0).
+
+    The bug shifts the bounds back 1h (2026-07-31 23:00 / 2026-08-31 23:00) and
+    then truncates each with ``.date()`` -> ``TO_DATE('2026-07-31')`` /
+    ``TO_DATE('2026-08-31')``. That window, [2026-07-31, 2026-08-31), admits the
+    out-of-range day 2026-07-31 and silently drops the last requested day,
+    2026-08-31.
+    """
+    sql = _generated_sql(_pg_dataset(1, "DATE"), mocker, app)
+
+    assert ">= TO_DATE('2026-08-01'" in sql, sql
+    assert "< TO_DATE('2026-09-01'" in sql, sql
+    # The lower bound must not admit the day before the requested range.
+    assert ">= TO_DATE('2026-07-31'" not in sql, (
+        f"DATE-column +1h offset admits out-of-range day 2026-07-31; SQL was:\n{sql}"
+    )
+    # The upper bound must not drop the last requested day (2026-08-31).
+    assert "< TO_DATE('2026-08-31'" not in sql, (
+        f"DATE-column +1h offset drops last requested day 2026-08-31; SQL was:\n{sql}"
+    )
+
+
+@pytest.mark.parametrize(
+    ("offset", "expected_start", "expected_end"),
+    [
+        (0, "2026-08-01", "2026-09-01"),
+        (1, "2026-08-01", "2026-09-01"),
+        (24, "2026-07-31", "2026-08-31"),
+        (25, "2026-07-31", "2026-08-31"),
+        (-1, "2026-08-01", "2026-09-01"),
+        (-25, "2026-08-02", "2026-09-02"),
+    ],
+)
+def test_date_column_hours_offset_uses_whole_day_bounds(
+    offset: int,
+    expected_start: str,
+    expected_end: str,
+    mocker: MockerFixture,
+    app: Flask,
+) -> None:
+    """DATE bounds discard sub-day remainders symmetrically around zero."""
+    sql = _generated_sql(_pg_dataset(offset, "DATE"), mocker, app)
+
+    assert f">= TO_DATE('{expected_start}'" in sql, sql
+    assert f"< TO_DATE('{expected_end}'" in sql, sql
+
+
+def test_timestamp_column_hours_offset_preserves_exact_hour_bounds(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """Control for Defect 1: the same +1h offset on a ``TIMESTAMP`` column keeps
+    exact-hour precision (2026-07-31 23:00:00 / 2026-08-31 23:00:00) and loses
+    nothing. This passes today and documents that the defect is DATE-specific."""
+    sql = _generated_sql(_pg_dataset(1, "TIMESTAMP"), mocker, app)
+
+    assert "2026-07-31 23:00:00" in sql, sql
+    assert "2026-08-31 23:00:00" in sql, sql
+
+
+@pytest.mark.parametrize(
+    ("offset", "expected_start", "expected_end"),
+    [
+        (0, "2026-08-01 00:00:00", "2026-09-01 00:00:00"),
+        (1, "2026-07-31 23:00:00", "2026-08-31 23:00:00"),
+        (24, "2026-07-31 00:00:00", "2026-08-31 00:00:00"),
+        (25, "2026-07-30 23:00:00", "2026-08-30 23:00:00"),
+        (-1, "2026-08-01 01:00:00", "2026-09-01 01:00:00"),
+        (-25, "2026-08-02 01:00:00", "2026-09-02 01:00:00"),
+    ],
+)
+def test_timestamp_column_hours_offset_uses_exact_hour_bounds(
+    offset: int,
+    expected_start: str,
+    expected_end: str,
+    mocker: MockerFixture,
+    app: Flask,
+) -> None:
+    """Timestamp bounds preserve every configured offset hour."""
+    sql = _generated_sql(_pg_dataset(offset, "TIMESTAMP"), mocker, app)
+
+    assert expected_start in sql, sql
+    assert expected_end in sql, sql
+
+
+def test_datetime_named_column_keeps_exact_hour_bounds(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """A DATETIME type name must not be mistaken for a pure DATE type."""
+    sql = _generated_sql(_pg_dataset(1, "DATETIME"), mocker, app)
+
+    assert "2026-07-31 23:00:00" in sql, sql
+    assert "2026-08-31 23:00:00" in sql, sql
+
+
+# ---------------------------------------------------------------------------
+# Defect 2 -- Hours offset applied after DB-side time-grain truncation
+# ---------------------------------------------------------------------------
+
+
+def _sqlite_dataset(
+    mocker: MockerFixture,
+    offset: int,
+    column_type: str,
+    rows: list[str],
+) -> tuple[SqlaTable, Engine]:
+    """Build an executable SQLite dataset with controlled temporal rows."""
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+        future=True,
+    )
+    database = Database(database_name="db", sqlalchemy_uri="sqlite://")
+    connection = engine.raw_connection()
+    connection.execute(f"CREATE TABLE events (ts {column_type}, val INTEGER)")
+    connection.executemany(
+        "INSERT INTO events VALUES (?, 1)",
+        [(row,) for row in rows],
+    )
+    connection.commit()
+
+    @contextmanager
+    def mock_get_sqla_engine(catalog=None, schema=None, **kwargs):
+        yield engine
+
+    mocker.patch.object(database, "get_sqla_engine", new=mock_get_sqla_engine)
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
+        return_value=[],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.is_guest_user",
+        return_value=False,
+    )
+
+    return (
+        SqlaTable(
+            database=database,
+            schema=None,
+            table_name="events",
+            main_dttm_col="ts",
+            offset=offset,
+            columns=[
+                TableColumn(column_name="ts", is_dttm=True, type=column_type),
+                TableColumn(column_name="val", type="INTEGER"),
+            ],
+        ),
+        engine,
+    )
+
+
+def _physical_axis_query(table: SqlaTable, time_grain: str | None) -> QueryObject:
+    """Build the physical-axis query shape used by legacy time-series charts."""
+    return QueryObject(
+        datasource=table,
+        metrics=[{"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "ct"}],
+        columns=[],
+        granularity="ts",
+        from_dttm=pd.Timestamp("2026-07-01"),
+        to_dttm=pd.Timestamp("2026-10-01"),
+        is_timeseries=True,
+        extras={"time_grain_sqla": time_grain} if time_grain else {},
+        row_limit=100,
+    )
+
+
+_EXPECTED_PHYSICAL_AXIS_TIMESTAMPS = {
+    ("TIMESTAMP", None): {
+        0: "2026-08-01 23:30:00",
+        1: "2026-08-02 00:30:00",
+        24: "2026-08-02 23:30:00",
+        25: "2026-08-03 00:30:00",
+        -1: "2026-08-01 22:30:00",
+        -25: "2026-07-31 22:30:00",
+    },
+    ("TIMESTAMP", "P1D"): {
+        0: "2026-08-01 00:00:00",
+        1: "2026-08-02 00:00:00",
+        24: "2026-08-02 00:00:00",
+        25: "2026-08-03 00:00:00",
+        -1: "2026-08-01 00:00:00",
+        -25: "2026-07-31 00:00:00",
+    },
+    ("DATE", None): {
+        0: "2026-08-02 00:00:00",
+        1: "2026-08-02 01:00:00",
+        24: "2026-08-03 00:00:00",
+        25: "2026-08-03 01:00:00",
+        -1: "2026-08-01 23:00:00",
+        -25: "2026-07-31 23:00:00",
+    },
+    ("DATE", "P1D"): {
+        0: "2026-08-02 00:00:00",
+        1: "2026-08-02 00:00:00",
+        24: "2026-08-03 00:00:00",
+        25: "2026-08-03 00:00:00",
+        -1: "2026-08-02 00:00:00",
+        -25: "2026-08-01 00:00:00",
+    },
+}
+
+
+@pytest.mark.parametrize("column_type", ["DATE", "TIMESTAMP"])
+@pytest.mark.parametrize("time_grain", [None, "P1D"])
+@pytest.mark.parametrize("offset", [0, 1, 24, 25, -1, -25])
+def test_physical_axis_offset_matrix(
+    column_type: str,
+    time_grain: str | None,
+    offset: int,
+    mocker: MockerFixture,
+) -> None:
+    """Physical axes apply each offset once at the precision of their grain."""
+    raw_value = "2026-08-02" if column_type == "DATE" else "2026-08-01 23:30:00"
+    table, _engine = _sqlite_dataset(mocker, offset, column_type, [raw_value])
+
+    result = table.get_query_result(_physical_axis_query(table, time_grain))
+
+    assert result.df["__timestamp"].tolist() == [
+        pd.Timestamp(
+            _EXPECTED_PHYSICAL_AXIS_TIMESTAMPS[(column_type, time_grain)][offset]
+        )
+    ]
+    expected_shifted_labels = {"__timestamp"} if time_grain and offset else set()
+    assert result.sql_shifted_temporal_labels == expected_shifted_labels
+
+
+def test_ungrained_physical_axis_offset_is_applied_exactly_once(
+    mocker: MockerFixture,
+) -> None:
+    """An ungrained axis stays on the established pandas-only offset path."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=1,
+        column_type="TIMESTAMP",
+        rows=["2026-08-01 23:30:00"],
+    )
+
+    result = table.get_query_result(_physical_axis_query(table, time_grain=None))
+
+    assert result.df["__timestamp"].tolist() == [pd.Timestamp("2026-08-02 00:30:00")]
+    assert result.sql_shifted_temporal_labels == set()
+
+
+def test_negative_subday_date_offset_does_not_move_grained_bucket(
+    mocker: MockerFixture,
+) -> None:
+    """A negative sub-day offset on a DATE grain quantizes to zero days."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=-1,
+        column_type="DATE",
+        rows=["2026-08-02"],
+    )
+
+    result = table.get_query_result(_physical_axis_query(table, time_grain="P1D"))
+
+    assert result.df["__timestamp"].tolist() == [pd.Timestamp("2026-08-02 00:00:00")]
+    assert result.sql_shifted_temporal_labels == {"__timestamp"}
+    assert "+0 hours" not in result.query
+
+
+def test_adhoc_base_axis_offset_is_applied_exactly_once(
+    mocker: MockerFixture,
+) -> None:
+    """A non-timeseries BASE_AXIS query shifts before its embedded grain."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=1,
+        column_type="TIMESTAMP",
+        rows=["2026-08-01 23:30:00", "2026-08-02 10:00:00"],
+    )
+    base_axis: AdhocColumn = {
+        "sqlExpression": "ts",
+        "label": "ts",
+        "isColumnReference": True,
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1D",
+    }
+    query_object = QueryObject(
+        datasource=table,
+        metrics=[{"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "ct"}],
+        columns=[base_axis],
+        granularity=None,
+        is_timeseries=False,
+        extras={},
+        row_limit=100,
+    )
+
+    result = table.get_query_result(query_object)
+
+    assert result.df["ts"].tolist() == [pd.Timestamp("2026-08-02 00:00:00")]
+    assert result.df["ct"].tolist() == [2]
+    assert result.sql_shifted_temporal_labels == {"ts"}
+
+
+def test_adhoc_axis_without_temporal_shift_capability_uses_pandas_fallback(
+    mocker: MockerFixture,
+) -> None:
+    """An ungated engine leaves an adhoc axis shift to pandas."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=1,
+        column_type="TIMESTAMP",
+        rows=["2026-08-01 23:30:00", "2026-08-02 10:00:00"],
+    )
+    mocker.patch.object(SqliteEngineSpec, "supports_temporal_column_shift", False)
+    base_axis: AdhocColumn = {
+        "sqlExpression": "ts",
+        "label": "ts",
+        "isColumnReference": True,
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1D",
+    }
+    query_object = QueryObject(
+        datasource=table,
+        metrics=[{"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "ct"}],
+        columns=[base_axis],
+        granularity=None,
+        is_timeseries=False,
+        extras={},
+        row_limit=100,
+    )
+
+    result = table.get_query_result(query_object)
+
+    assert set(result.df["ts"]) == {
+        pd.Timestamp("2026-08-01 01:00:00"),
+        pd.Timestamp("2026-08-02 01:00:00"),
+    }
+    assert result.df["ct"].tolist() == [1, 1]
+    assert result.sql_shifted_temporal_labels == set()
+    assert "+1 hours" not in result.query
+    assert "DATETIME(DATETIME(ts" not in result.query
+
+
+def test_adhoc_base_axis_probe_quantizes_date_offset(
+    mocker: MockerFixture,
+) -> None:
+    """A probed DATE expression quantizes a sub-day offset to zero hours."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=-1,
+        column_type="TIMESTAMP",
+        rows=["2026-08-02"],
+    )
+    probe = mocker.patch(
+        "superset.connectors.sqla.models.get_columns_description",
+        return_value=[{"is_dttm": True, "type": "DATE"}],
+    )
+    base_axis: AdhocColumn = {
+        "sqlExpression": "DATE(ts)",
+        "label": "ts",
+        "isColumnReference": False,
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1D",
+    }
+    query_object = QueryObject(
+        datasource=table,
+        metrics=[{"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "ct"}],
+        columns=[base_axis],
+        granularity=None,
+        is_timeseries=False,
+        extras={},
+        row_limit=100,
+    )
+
+    result = table.get_query_result(query_object)
+
+    probe.assert_called()
+    assert result.df["ts"].tolist() == [pd.Timestamp("2026-08-02 00:00:00")]
+    assert result.sql_shifted_temporal_labels == {"ts"}
+    assert "-1 hours" not in result.query
+    assert "DATETIME(DATETIME(DATE(ts)" not in result.query
+
+
+def test_engine_without_temporal_shift_capability_uses_pandas_fallback(
+    mocker: MockerFixture,
+) -> None:
+    """An ungated engine leaves the axis unshifted and applies the offset in pandas."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=1,
+        column_type="TIMESTAMP",
+        rows=["2026-08-01 23:30:00"],
+    )
+    mocker.patch.object(SqliteEngineSpec, "supports_temporal_column_shift", False)
+
+    result = table.get_query_result(_physical_axis_query(table, time_grain="P1D"))
+
+    assert result.df["__timestamp"].tolist() == [pd.Timestamp("2026-08-01 01:00:00")]
+    assert result.sql_shifted_temporal_labels == set()
+    assert "+1 hours" not in result.query
+    assert "DATETIME(DATETIME(ts" not in result.query
+
+
+def test_grained_physical_filter_sql_is_unchanged(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """Physical grained filters keep their pre-existing unshifted expression."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=1,
+        column_type="TIMESTAMP",
+        rows=[],
+    )
+    query_object = QueryObject(
+        datasource=table,
+        columns=["val"],
+        metrics=[],
+        is_timeseries=False,
+        filters=[
+            {
+                "col": "ts",
+                "op": "TEMPORAL_RANGE",
+                "val": "2026-08-02 : 2026-08-03",
+                "grain": "P1D",
+            }
+        ],
+    )
+    with app.test_request_context():
+        query = table.get_query_str_extended(query_object.to_dict(), mutate=False)
+
+    assert query.sql == (
+        "SELECT val AS val \n"
+        "FROM events \n"
+        "WHERE DATETIME(ts, 'start of day') >= '2026-08-01 23:00:00' "
+        "AND DATETIME(ts, 'start of day') < '2026-08-02 23:00:00' GROUP BY val"
+    )
+    assert query.sql_shifted_temporal_labels == set()
+
+
+def test_grained_adhoc_filter_sql_is_unchanged(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """A BASE_AXIS-shaped adhoc filter does not opt into the axis-only shift."""
+    table, _engine = _sqlite_dataset(
+        mocker,
+        offset=1,
+        column_type="TIMESTAMP",
+        rows=[],
+    )
+    base_axis_filter: AdhocColumn = {
+        "sqlExpression": "ts",
+        "label": "ts",
+        "isColumnReference": True,
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1D",
+    }
+    query_object = QueryObject(
+        datasource=table,
+        columns=["val"],
+        metrics=[],
+        is_timeseries=False,
+        filters=[
+            {
+                "col": base_axis_filter,
+                "op": "==",
+                "val": "2026-08-02 00:00:00",
+            }
+        ],
+    )
+    with app.test_request_context():
+        query = table.get_query_str_extended(query_object.to_dict(), mutate=False)
+
+    assert query.sql == (
+        "SELECT val AS val \n"
+        "FROM events \n"
+        "WHERE (DATETIME(ts, 'start of day')) = '2026-08-02 00:00:00' GROUP BY val"
+    )
+    assert query.sql_shifted_temporal_labels == set()
+
+
+@pytest.mark.parametrize(
+    ("offset", "postgres_sql", "sqlite_sql"),
+    [
+        (1, "ts + INTERVAL '1' HOUR", "DATETIME(ts, '+1 hours')"),
+        (-1, "ts + INTERVAL '-1' HOUR", "DATETIME(ts, '-1 hours')"),
+        (24, "ts + INTERVAL '24' HOUR", "DATETIME(ts, '+24 hours')"),
+        (-25, "ts + INTERVAL '-25' HOUR", "DATETIME(ts, '-25 hours')"),
+    ],
+)
+def test_temporal_column_shift_expression_compiles_for_supported_guard_dialects(
+    offset: int,
+    postgres_sql: str,
+    sqlite_sql: str,
+) -> None:
+    """The bounded engine hook emits valid PostgreSQL and SQLite shift syntax."""
+    source = column("ts", type_=DateTime())
+
+    postgres_shift = PostgresEngineSpec.get_temporal_column_shift_expr(source, offset)
+    sqlite_shift = SqliteEngineSpec.get_temporal_column_shift_expr(source, offset)
+    postgres_bucket = PostgresEngineSpec.get_timestamp_expr(postgres_shift, None, "P1D")
+    sqlite_bucket = SqliteEngineSpec.get_timestamp_expr(sqlite_shift, None, "P1D")
+
+    assert BaseEngineSpec.supports_temporal_column_shift is False
+    assert PostgresEngineSpec.supports_temporal_column_shift is True
+    assert SqliteEngineSpec.supports_temporal_column_shift is True
+    assert str(postgres_shift.compile(dialect=postgresql.dialect())) == postgres_sql
+    assert str(sqlite_shift.compile(dialect=sqlite.dialect())) == sqlite_sql
+    assert str(postgres_bucket.compile(dialect=postgresql.dialect())) == (
+        f"DATE_TRUNC('day', {postgres_sql})"
+    )
+    assert str(sqlite_bucket.compile(dialect=sqlite.dialect())) == (
+        f"DATETIME({sqlite_sql}, 'start of day')"
+    )
+
+
+def test_hours_offset_is_applied_before_time_grain_truncation(
+    mocker: MockerFixture, session: Session
+) -> None:
+    """A row near a day boundary must be bucketed by the grain using its
+    offset-shifted (local) time, not its raw time.
+
+    Raw ``2026-08-01 23:30`` at a +1h dataset offset is locally ``2026-08-02
+    00:30``; under a daily grain it belongs to 2026-08-02. The bug truncates the
+    raw value to 2026-08-01 in the database and only then adds the offset in
+    pandas, so the row is mislabeled as 2026-08-01 (a full day early).
+    """
+    SqlaTable.metadata.create_all(session.get_bind())
+
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+        future=True,
+    )
+    database = Database(database_name="db", sqlalchemy_uri="sqlite://")
+    connection = engine.raw_connection()
+    connection.execute("CREATE TABLE events (ts TIMESTAMP, val INTEGER)")
+    # Boundary row (local day 2026-08-02) and a same-local-day daytime row.
+    connection.execute("INSERT INTO events VALUES ('2026-08-01 23:30:00', 1)")
+    connection.execute("INSERT INTO events VALUES ('2026-08-02 10:00:00', 1)")
+    connection.commit()
+
+    @contextmanager
+    def mock_get_sqla_engine(catalog=None, schema=None, **kwargs):
+        yield engine
+
+    mocker.patch.object(database, "get_sqla_engine", new=mock_get_sqla_engine)
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
+        return_value=[],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.is_guest_user",
+        return_value=False,
+    )
+
+    table = SqlaTable(
+        database=database,
+        schema=None,
+        table_name="events",
+        main_dttm_col="ts",
+        offset=1,
+        columns=[
+            TableColumn(column_name="ts", is_dttm=True, type="TIMESTAMP"),
+            TableColumn(column_name="val", type="INTEGER"),
+        ],
+    )
+
+    from superset.common.query_object import QueryObject
+
+    query_object = QueryObject(
+        datasource=table,
+        metrics=[{"expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "ct"}],
+        columns=[],
+        granularity="ts",
+        from_dttm=pd.Timestamp("2026-07-01"),
+        to_dttm=pd.Timestamp("2026-10-01"),
+        is_timeseries=True,
+        extras={"time_grain_sqla": "P1D"},
+        filters=[
+            {"col": "ts", "op": "TEMPORAL_RANGE", "val": "2026-07-01 : 2026-10-01"}
+        ],
+        row_limit=100,
+    )
+
+    result = table.get_query_result(query_object)
+    bucket_days = {ts.date() for ts in result.df["__timestamp"]}
+
+    # Both rows are locally on 2026-08-02, so every bucket must be 2026-08-02.
+    # The bug leaves the boundary row on 2026-08-01.
+    assert date(2026, 8, 1) not in bucket_days, (
+        "Row raw 2026-08-01 23:30 (local 2026-08-02 00:30) was bucketed to "
+        f"2026-08-01, a day early. Buckets: {sorted(bucket_days)}"
+    )
+    assert bucket_days == {date(2026, 8, 2)}, (
+        f"All rows should bucket to 2026-08-02; got {sorted(bucket_days)}"
+    )