feat(datasets): add partition filter mapping model, migration and validation

Datasets on Hadoop-family engines are often partitioned on a technical column
-- an epoch integer, a lowercased region key -- that no analyst would filter
on. Unless a query carries a predicate on that column the engine scans every
partition, and today the only workaround is hand-writing the predicate as
custom SQL in a virtual dataset.

This is the first of four PRs making that a dataset setting. It adds the
storage and the save-time validation; nothing reads the mapping yet.

Four columns, following the `always_filter_main_dttm` / `currency_code_column`
precedent for "a dataset-level setting that names a column":

  tables.partition_column                        the physical partition column
  tables.partition_mapped_column                 override; NULL follows main_dttm_col
  table_columns.partition_value_transform        the `:value` expression
  table_columns.partition_transform_is_monotonic gates range mirroring

The monotonic flag is NOT NULL DEFAULT false rather than a nullable tri-state,
matching `normalize_columns` -- a nullable boolean invites `if x:` bugs where
None and False need distinguishing and don't get it.

Validation runs in two tiers. Structural and safety errors block the save:
unknown columns, a column mapped onto itself, Jinja in the transform, and
non-deterministic functions. Everything else -- an unparseable transform, a
transform missing `:value` -- saves and leaves the mapping inactive, so a
half-written transform doesn't cost the owner the rest of their edits.

Note the self-mapping check validates the *effective* mapped column. Checking
only the explicit override misses the case an owner actually hits: pointing
`partition_column` at the column that is already `main_dttm_col`.

`SQLStatement.get_niladic_functions` is added because the denylist cannot be
purely name-based: on Hive and Impala `unix_timestamp()` means "now" while
`unix_timestamp(x)` -- the canonical transform for this feature -- is pure.

Gated behind the `PARTITION_FILTER_MAPPING` feature flag, off by default.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/static/feature-flags.json

@@ -81,6 +81,12 @@
         "lifecycle": "development",
         "description": "Try to optimize SQL queries \u2014 for now only predicate pushdown is supported"
       },
+      {
+        "name": "PARTITION_FILTER_MAPPING",
+        "default": false,
+        "lifecycle": "development",
+        "description": "Mirror filters on a dataset's business column onto its physical partition column, so engines that require an explicit partition predicate can prune."
+      },
       {
         "name": "PRESTO_EXPAND_DATA",
         "default": false,

docs/static/resources/openapi.json

@@ -3897,6 +3897,16 @@
           "params": {
             "type": "string"
           },
+          "partition_column": {
+            "type": "string"
+          },
+          "partition_filter_mapping": {
+            "nullable": true,
+            "type": "object"
+          },
+          "partition_mapped_column": {
+            "type": "string"
+          },
           "perm": {
             "type": "string"
           },
@@ -5788,6 +5798,15 @@
             "nullable": true,
             "type": "boolean"
           },
+          "partition_transform_is_monotonic": {
+            "default": false,
+            "type": "boolean"
+          },
+          "partition_value_transform": {
+            "description": "SQL expression containing a :value placeholder. Filters on this column are mirrored onto the dataset's partition column with the value passed through this transform.",
+            "nullable": true,
+            "type": "string"
+          },
           "python_date_format": {
             "maxLength": 255,
             "minLength": 1,
@@ -6253,6 +6272,16 @@
           "order_by_choices": {
             "readOnly": true
           },
+          "partition_column": {
+            "maxLength": 250,
+            "nullable": true,
+            "type": "string"
+          },
+          "partition_mapped_column": {
+            "maxLength": 250,
+            "nullable": true,
+            "type": "string"
+          },
           "schema": {
             "maxLength": 255,
             "nullable": true,
@@ -6721,6 +6750,18 @@
             "default": false,
             "type": "boolean"
           },
+          "partition_column": {
+            "maxLength": 250,
+            "minLength": 0,
+            "nullable": true,
+            "type": "string"
+          },
+          "partition_mapped_column": {
+            "maxLength": 250,
+            "minLength": 0,
+            "nullable": true,
+            "type": "string"
+          },
           "schema": {
             "maxLength": 250,
             "minLength": 0,
@@ -6847,6 +6888,18 @@
             "nullable": true,
             "type": "integer"
           },
+          "partition_column": {
+            "maxLength": 250,
+            "minLength": 0,
+            "nullable": true,
+            "type": "string"
+          },
+          "partition_mapped_column": {
+            "maxLength": 250,
+            "minLength": 0,
+            "nullable": true,
+            "type": "string"
+          },
           "schema": {
             "maxLength": 255,
             "minLength": 0,

superset/commands/dataset/update.py

@@ -46,6 +46,10 @@ from superset.commands.dataset.exceptions import (
 )
 from superset.commands.utils import compute_subjects
 from superset.connectors.sqla.models import SqlaTable, validate_stored_expression
+from superset.connectors.sqla.partition_mapping import (
+    parse_skeleton,
+    validate_partition_mapping,
+)
 from superset.daos.dataset import DatasetDAO
 from superset.datasets.schemas import FolderSchema
 from superset.exceptions import (
@@ -281,6 +285,8 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
         if predicate := self._properties.get("fetch_values_predicate"):
             self._validate_fetch_values_predicate(predicate, exceptions)
 
+        self._validate_partition_mapping(exceptions)
+
         if folders := self._properties.get("folders"):
             valid_uuids: set[UUID] = set()
             if metrics:
@@ -389,6 +395,103 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
                     )
                 )
 
+    def _validate_partition_mapping(self, exceptions: list[ValidationError]) -> None:
+        """
+        Validate the dataset's partition filter mapping.
+
+        Only the blocking (Tier 1) issues become validation errors. Tier 2
+        issues -- an unparseable transform, a transform missing `:value` --
+        deliberately let the save through and leave the mapping inactive, per
+        the PRD, so a half-written transform doesn't cost the owner the rest of
+        their edits. They are surfaced by the editor, not by rejecting the PUT.
+
+        The transform is authored by a dataset owner, the same principal and
+        trust level as a calculated-column expression, so it also goes through
+        `validate_stored_expression` -- the parser gate that already governs
+        stored expressions.
+        """
+        self._model = cast(SqlaTable, self._model)
+
+        columns = self._properties.get("columns")
+        column_names = (
+            {column["column_name"] for column in columns}
+            if columns is not None
+            else {column.column_name for column in self._model.columns}
+        )
+
+        partition_column = self._properties.get(
+            "partition_column", self._model.partition_column
+        )
+        partition_mapped_column = self._properties.get(
+            "partition_mapped_column", self._model.partition_mapped_column
+        )
+        main_dttm_col = self._properties.get("main_dttm_col", self._model.main_dttm_col)
+        if not partition_column:
+            return
+
+        database = self._properties.get("database") or self._model.database
+        catalog = self._properties.get("catalog", self._model.catalog)
+        schema = self._properties.get("schema", self._model.schema)
+
+        effective_mapped_column = partition_mapped_column or main_dttm_col
+        transform = self._effective_transform(columns, effective_mapped_column)
+
+        for issue in validate_partition_mapping(
+            column_names=column_names,
+            partition_column=partition_column,
+            partition_mapped_column=partition_mapped_column,
+            main_dttm_col=main_dttm_col,
+            transform=transform,
+            engine=database.backend,
+        ):
+            if issue.blocking:
+                exceptions.append(
+                    ValidationError(str(issue.message), field_name=issue.field)
+                )
+
+        if transform:
+            try:
+                validate_stored_expression(
+                    database, catalog, schema, parse_skeleton(transform)
+                )
+            except (SupersetSecurityException, QueryClauseValidationException) as ex:
+                message = (
+                    ex.error.message
+                    if isinstance(ex, SupersetSecurityException)
+                    else ex.message
+                )
+                exceptions.append(
+                    ValidationError(
+                        message,
+                        field_name="partition_value_transform",
+                    )
+                )
+
+    def _effective_transform(
+        self,
+        columns: list[dict[str, Any]] | None,
+        mapped_column: str | None,
+    ) -> str | None:
+        """
+        The value transform on the effective mapped column.
+
+        Reads from the payload when the request carries columns, and from the
+        persisted model otherwise -- a PUT that changes only `partition_column`
+        still has to be validated against the transform already stored.
+        """
+        if not mapped_column:
+            return None
+        if columns is not None:
+            for column in columns:
+                if column.get("column_name") == mapped_column:
+                    return column.get("partition_value_transform")
+            return None
+        model = cast(SqlaTable, self._model)
+        for existing in model.columns:
+            if existing.column_name == mapped_column:
+                return existing.partition_value_transform
+        return None
+
     def _validate_fetch_values_predicate(
         self,
         predicate: str,

superset/config.py

@@ -733,6 +733,10 @@ DEFAULT_FEATURE_FLAGS: dict[str, bool] = {
     # Try to optimize SQL queries — for now only predicate pushdown is supported
     # @lifecycle: development
     "OPTIMIZE_SQL": False,
+    # Mirror filters on a dataset's business column onto its physical partition
+    # column, so engines that require an explicit partition predicate can prune.
+    # @lifecycle: development
+    "PARTITION_FILTER_MAPPING": False,
     # Expand nested types in Presto into extra columns/arrays. Experimental,
     # doesn't work with all nested types.
     # @lifecycle: development

superset/connectors/sqla/models.py

@@ -1069,6 +1069,24 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
     python_date_format = Column(String(255))
     datetime_format = Column(String(100))
     extra = Column(Text)
+    # Partition filter mapping (§ PARTITION_FILTER_MAPPING). The transform is a
+    # SQL expression containing a `:value` placeholder; filters on this column
+    # are mirrored onto the dataset's `partition_column` as
+    # `partition_column <op> <transform evaluated at :value>`.
+    partition_value_transform = Column(Text)
+    # Whether the transform preserves ordering. Range operators (and time
+    # ranges) are only mirrored when it does; see the operator matrix in
+    # `superset.connectors.sqla.partition_mapping`.
+    #
+    # Nullable, like every other boolean on this model. The legacy datasource
+    # editor saves through `update_from_object`, which writes `obj.get(attr)`
+    # for every field in `update_from_object_fields` -- so any field its payload
+    # omits is written as NULL. A NOT NULL column here fails that save outright.
+    # Readers coerce with `bool(...)`, so NULL means "not declared", which is
+    # the safe direction: ranges stop mirroring rather than mirroring unsoundly.
+    partition_transform_is_monotonic = Column(
+        Boolean, default=False, server_default=sa.false()
+    )
 
     table: Mapped["SqlaTable"] = relationship(
         "SqlaTable",
@@ -1091,6 +1109,8 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
         "python_date_format",
         "datetime_format",
         "extra",
+        "partition_value_transform",
+        "partition_transform_is_monotonic",
     ]
 
     update_from_object_fields = [s for s in export_fields if s not in ("table_id",)]
@@ -1361,6 +1381,8 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
             "type_generic",
             "verbose_name",
             "warning_markdown",
+            "partition_value_transform",
+            "partition_transform_is_monotonic",
         )
 
         return {s: getattr(self, s) for s in attrs if hasattr(self, s)}
@@ -1595,6 +1617,13 @@ class SqlaTable(
     normalize_columns = Column(Boolean, default=False)
     always_filter_main_dttm = Column(Boolean, default=False)
     folders = Column(JSON, nullable=True)
+    # Physical column the engine partitions on. Filters on the effective mapped
+    # column are mirrored onto it so the engine can prune partitions.
+    partition_column = Column(String(250))
+    # Explicit override for the column whose filters are mirrored. NULL means
+    # "follow `main_dttm_col`", so re-pointing the default datetime column moves
+    # the mapping with it.
+    partition_mapped_column = Column(String(250))
 
     baselink = "tablemodelview"
 
@@ -1618,6 +1647,8 @@ class SqlaTable(
         "normalize_columns",
         "always_filter_main_dttm",
         "folders",
+        "partition_column",
+        "partition_mapped_column",
     ]
     update_from_object_fields = [f for f in export_fields if f != "database_id"]
     export_parent = "database"
@@ -1845,8 +1876,44 @@ class SqlaTable(
             data_["extra"] = self.extra
             data_["always_filter_main_dttm"] = self.always_filter_main_dttm
             data_["normalize_columns"] = self.normalize_columns
+            data_["partition_column"] = self.partition_column
+            data_["partition_mapped_column"] = self.partition_mapped_column
+            data_["partition_filter_mapping"] = self.partition_filter_mapping_summary
         return data_
 
+    @property
+    def partition_filter_mapping_summary(self) -> dict[str, Any] | None:
+        """
+        Self-contained summary of the mapping for the Explore indicator.
+
+        Deliberately not a lookup into `columns`: `data_for_slices` prunes
+        columns no chart references, and the partition column is typically
+        referenced by none of them, so anything reading it out of
+        `datasource.columns` would work in Explore and break on dashboards.
+
+        `active` is derived from cheap signals only. This property is serialized
+        on every chart and dashboard load, so parsing the transform here would
+        put a per-request cost on a hot path for a value that only changes on
+        save.
+        """
+        if not self.partition_column:
+            return None
+
+        columns_by_name = {column.column_name: column for column in self.columns}
+        mapped_column_name = self.partition_mapped_column or self.main_dttm_col
+        mapped_column = columns_by_name.get(mapped_column_name or "")
+        active = bool(
+            self.partition_column in columns_by_name
+            and mapped_column is not None
+            and mapped_column_name != self.partition_column
+            and (mapped_column.partition_value_transform or "").strip()
+        )
+        return {
+            "partition_column": self.partition_column,
+            "mapped_column": mapped_column_name,
+            "active": active,
+        }
+
     @property
     def extra_dict(self) -> dict[str, Any]:
         try:

superset/connectors/sqla/partition_mapping.py

@@ -0,0 +1,324 @@
+# 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.
+"""
+Partition filter mapping.
+
+Datasets on Hadoop-family engines are commonly partitioned on a *technical*
+column -- an epoch integer, a lowercased region key -- that no analyst would
+filter on. Unless a query carries a predicate on that column the engine scans
+every partition.
+
+A dataset owner names one partition column ``p``, one business column that
+filters are mirrored from, and a value transform ``T`` (a SQL expression
+containing a ``:value`` placeholder). Superset then appends an equivalent
+predicate on ``p`` to every query, so chart authors change nothing and queries
+prune.
+
+The load-bearing assumption
+---------------------------
+Everything here reasons about ``T(col) op T(v)``, but what is emitted is
+``p op T(v)`` -- a predicate on a *physically different column*. The step from
+one to the other is::
+
+    p = T(mapped_col)   for every row in the table
+
+Superset cannot verify that; it is a property of whatever ETL populates the
+partition column. If that job lags, backfills with different logic, or writes
+the partition key in a different timezone, mirrored predicates silently drop
+real rows. The mapping is only as trustworthy as the pipeline behind it.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+
+from flask_babel import lazy_gettext as _
+
+from superset.exceptions import SupersetParseError
+from superset.sql.parse import SQLStatement
+
+logger = logging.getLogger(__name__)
+
+FEATURE_FLAG = "PARTITION_FILTER_MAPPING"
+
+#: Placeholder the owner writes in the transform, e.g. ``unix_timestamp(:value)``.
+#: Matched with word boundaries so ``:values`` is not mistaken for it.
+VALUE_PLACEHOLDER_RE = re.compile(r":value\b")
+
+#: Balanced Jinja blocks. The probe would render these in a different context
+#: at a different time from the chart query, so they are rejected at save time.
+JINJA_BLOCK_RE = re.compile(r"\{\{.*?\}\}|\{%.*?%\}|\{#.*?#\}", re.DOTALL)
+
+#: Substituted for ``:value`` before parsing -- sqlglot rejects a bare ``:value``
+#: on most dialects. Mirrors the ``_JINJA_BLOCK_RE`` -> ``NULL`` trick used by
+#: ``validate_stored_expression``.
+_PARSE_STANDIN = "NULL"
+
+#: Functions whose value depends on wall-clock time or randomness. The probe
+#: runs in a different session at a different moment from the chart query and
+#: its result is then cached, so any of these freezes a snapshot of probe time
+#: into the emitted predicate.
+NON_DETERMINISTIC_FUNCTIONS = {
+    "CURRENT_DATE",
+    "CURRENT_TIME",
+    "CURRENT_TIMESTAMP",
+    "NOW",
+    "RAND",
+    "RANDOM",
+    "UUID",
+}
+
+#: Functions that mean "now" only in their zero-argument form. On Hive and
+#: Impala ``unix_timestamp()`` is the current time while ``unix_timestamp(x)``
+#: -- the canonical transform for this feature -- is pure.
+NON_DETERMINISTIC_WHEN_NILADIC = {"UNIX_TIMESTAMP"}
+
+
+def contains_value_placeholder(transform: str | None) -> bool:
+    """Whether the transform contains the ``:value`` placeholder."""
+    return bool(transform) and VALUE_PLACEHOLDER_RE.search(transform or "") is not None
+
+
+def contains_jinja(transform: str | None) -> bool:
+    """Whether the transform contains a balanced Jinja block."""
+    return bool(transform) and JINJA_BLOCK_RE.search(transform or "") is not None
+
+
+def parse_skeleton(transform: str) -> str:
+    """
+    The transform with ``:value`` substituted out, ready for a SQL parser.
+
+    ``sanitize_clause`` / sqlglot choke on a bare ``:value`` on most dialects,
+    so the placeholder is swapped for a benign literal first -- the same trick
+    ``validate_stored_expression`` uses for Jinja blocks.
+    """
+    return VALUE_PLACEHOLDER_RE.sub(_PARSE_STANDIN, transform)
+
+
+def _parse_skeleton(transform: str, engine: str) -> SQLStatement | None:
+    """
+    Parse ``SELECT <transform>`` with the placeholder substituted out.
+
+    Returns ``None`` when the transform does not parse.
+    """
+    try:
+        return SQLStatement(f"SELECT {parse_skeleton(transform)}", engine)
+    except SupersetParseError:
+        return None
+
+
+def is_parseable(transform: str | None, engine: str) -> bool:
+    """Whether the transform parses as a single select expression."""
+    if not transform or not transform.strip():
+        return False
+    return _parse_skeleton(transform, engine) is not None
+
+
+def find_non_deterministic_functions(transform: str, engine: str) -> set[str]:
+    """
+    Names of non-deterministic functions the transform calls.
+
+    ``UNIX_TIMESTAMP`` is only reported in its zero-argument form, which means
+    "now" on Hive and Impala; the one-argument form is the canonical temporal
+    transform and stays allowed.
+    """
+    statement = _parse_skeleton(transform, engine)
+    if statement is None:
+        return set()
+
+    found = {
+        name
+        for name in NON_DETERMINISTIC_FUNCTIONS
+        if statement.check_functions_present({name})
+    }
+    return found | _find_niladic_calls(statement)
+
+
+def _find_niladic_calls(statement: SQLStatement) -> set[str]:
+    """
+    Names from ``NON_DETERMINISTIC_WHEN_NILADIC`` called with no arguments.
+
+    Note some dialects resolve the zero-argument form themselves -- Hive parses
+    ``unix_timestamp()`` straight to ``CURRENT_TIMESTAMP`` -- in which case the
+    name-based check above has already caught it. This is the backstop for the
+    dialects that do not.
+    """
+    return NON_DETERMINISTIC_WHEN_NILADIC & statement.get_niladic_functions()
+
+
+@dataclass(frozen=True)
+class MappingValidationIssue:
+    """
+    One problem found with a mapping at save time.
+
+    ``blocking`` issues reject the save (400). The rest save fine and leave the
+    mapping inactive -- the PRD is explicit that a mapping "stays inactive until
+    it parses", so a half-written transform must not cost the owner the rest of
+    their edits.
+    """
+
+    field: str
+    message: str
+    blocking: bool
+
+
+def validate_partition_mapping(  # pylint: disable=too-many-arguments
+    *,
+    column_names: set[str],
+    partition_column: str | None,
+    partition_mapped_column: str | None,
+    main_dttm_col: str | None,
+    transform: str | None,
+    engine: str,
+) -> list[MappingValidationIssue]:
+    """
+    Validate a dataset's partition mapping, in two tiers.
+
+    Tier 1 (``blocking=True``) is structural and safety: the columns have to
+    exist, a column cannot be mapped onto itself, and the transform cannot carry
+    Jinja or call a non-deterministic function. Tier 2 (``blocking=False``) is
+    everything that merely leaves the mapping inactive.
+
+    The Tier-1 transform checks need a successful parse to inspect anything, so
+    an unparseable transform falls through to Tier 2. That leaves them
+    unreachable in exactly the case where it doesn't matter: an unparseable
+    transform is never executed.
+    """
+    if not partition_column:
+        return []
+
+    issues: list[MappingValidationIssue] = []
+
+    if partition_column not in column_names:
+        issues.append(
+            MappingValidationIssue(
+                field="partition_column",
+                message=_(
+                    "Partition column %(name)s is not a column on this dataset.",
+                    name=partition_column,
+                ),
+                blocking=True,
+            )
+        )
+
+    if partition_mapped_column and partition_mapped_column not in column_names:
+        issues.append(
+            MappingValidationIssue(
+                field="partition_mapped_column",
+                message=_(
+                    "Mapped column %(name)s is not a column on this dataset.",
+                    name=partition_mapped_column,
+                ),
+                blocking=True,
+            )
+        )
+
+    effective_mapped_column = partition_mapped_column or main_dttm_col
+    if effective_mapped_column and effective_mapped_column == partition_column:
+        issues.append(
+            MappingValidationIssue(
+                field="partition_column",
+                message=_(
+                    "The partition column cannot be mapped onto itself. "
+                    "%(name)s is both the partition column and the mapped "
+                    "column.",
+                    name=partition_column,
+                ),
+                blocking=True,
+            )
+        )
+
+    issues.extend(validate_transform(transform, engine))
+    return issues
+
+
+def validate_transform(
+    transform: str | None,
+    engine: str,
+) -> list[MappingValidationIssue]:
+    """Validate the value transform on its own. See `validate_partition_mapping`."""
+    field = "partition_value_transform"
+
+    if contains_jinja(transform):
+        return [
+            MappingValidationIssue(
+                field=field,
+                message=_(
+                    "Jinja templating is not supported in a partition value "
+                    "transform. The transform is evaluated in a different "
+                    "context and at a different time from the chart query, so "
+                    "a template would not render the same way."
+                ),
+                blocking=True,
+            )
+        ]
+
+    if not transform or not transform.strip():
+        return [
+            MappingValidationIssue(
+                field=field,
+                message=_(
+                    "No value transform is set, so no filter will be mirrored "
+                    "onto the partition column."
+                ),
+                blocking=False,
+            )
+        ]
+
+    if not is_parseable(transform, engine):
+        return [
+            MappingValidationIssue(
+                field=field,
+                message=_(
+                    "The value transform could not be parsed. The mapping is "
+                    "saved but stays inactive until it does."
+                ),
+                blocking=False,
+            )
+        ]
+
+    if not contains_value_placeholder(transform):
+        return [
+            MappingValidationIssue(
+                field=field,
+                message=_(
+                    "The value transform must contain the :value placeholder, "
+                    "which stands for the filter value being mirrored."
+                ),
+                blocking=False,
+            )
+        ]
+
+    if functions := find_non_deterministic_functions(transform, engine):
+        return [
+            MappingValidationIssue(
+                field=field,
+                message=_(
+                    "The value transform calls %(functions)s, whose result "
+                    "depends on when and where it runs. The transform is "
+                    "evaluated in a separate session and the result is cached, "
+                    "so the emitted predicate would freeze a snapshot of that "
+                    "moment.",
+                    functions=", ".join(sorted(functions)),
+                ),
+                blocking=True,
+            )
+        ]
+
+    return []

superset/daos/dataset.py

@@ -448,6 +448,37 @@ class DatasetDAO(BaseDAO[SqlaTable]):
                     "python_date_format is an invalid date/timestamp format."
                 )
 
+    @staticmethod
+    def clear_dangling_partition_mapping(
+        model: SqlaTable, surviving_column_names: set[str]
+    ) -> None:
+        """
+        Drop parts of the partition filter mapping whose columns no longer exist.
+
+        A metadata sync can remove the partition column at the source, which
+        would otherwise leave the dataset pointing at a column that isn't there.
+        The query layer bails out defensively on a dangling mapping, so this is
+        about the dataset's stored state being honest rather than about
+        correctness of the SQL.
+
+        Called from the backend `override_columns=true` path, which is the
+        authoritative one: the editor clears the mapping client-side too, but an
+        API-driven sync bypasses the editor entirely.
+        """
+        if (
+            model.partition_column
+            and model.partition_column not in surviving_column_names
+        ):
+            model.partition_column = None
+            model.partition_mapped_column = None
+            return
+
+        if (
+            model.partition_mapped_column
+            and model.partition_mapped_column not in surviving_column_names
+        ):
+            model.partition_mapped_column = None
+
     @classmethod
     def _override_columns(
         cls, model: SqlaTable, property_columns: list[dict[str, Any]]
@@ -518,6 +549,8 @@ class DatasetDAO(BaseDAO[SqlaTable]):
                 }
                 db.session.add(TableColumn(**{**cleaned, "table_id": model.id}))
 
+        cls.clear_dangling_partition_mapping(model, set(incoming_by_name))
+
     @classmethod
     def _upsert_columns(
         cls, model: SqlaTable, property_columns: list[dict[str, Any]]

superset/dashboards/schemas.py

@@ -372,6 +372,9 @@ class DashboardDatasetSchema(Schema):
     granularity_sqla = fields.List(fields.List(fields.Str()))
     normalize_columns = fields.Bool()
     always_filter_main_dttm = fields.Bool()
+    partition_column = fields.Str()
+    partition_mapped_column = fields.Str()
+    partition_filter_mapping = fields.Dict(allow_none=True)
 
     # pylint: disable=unused-argument
     @post_dump()

superset/datasets/api.py

@@ -242,6 +242,8 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
         "description",
         "main_dttm_col",
         "currency_code_column",
+        "partition_column",
+        "partition_mapped_column",
         "normalize_columns",
         "always_filter_main_dttm",
         "offset",
@@ -345,6 +347,8 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
         "description",
         "main_dttm_col",
         "currency_code_column",
+        "partition_column",
+        "partition_mapped_column",
         "normalize_columns",
         "always_filter_main_dttm",
         "offset",

superset/datasets/schemas.py

@@ -99,6 +99,17 @@ class DatasetColumnsPutSchema(Schema):
     datetime_format = fields.String(
         allow_none=True, validate=[Length(1, 100), validate_python_date_format]
     )
+    partition_value_transform = fields.String(
+        allow_none=True,
+        metadata={
+            "description": (
+                "SQL expression containing a :value placeholder. Filters on "
+                "this column are mirrored onto the dataset's partition column "
+                "with the value passed through this transform."
+            )
+        },
+    )
+    partition_transform_is_monotonic = fields.Boolean(load_default=False)
     uuid = fields.UUID(allow_none=True)
 
 
@@ -177,6 +188,8 @@ class DatasetPostSchema(Schema):
     normalize_columns = fields.Boolean(load_default=False)
     always_filter_main_dttm = fields.Boolean(load_default=False)
     currency_code_column = fields.String(allow_none=True, validate=Length(0, 250))
+    partition_column = fields.String(allow_none=True, validate=Length(0, 250))
+    partition_mapped_column = fields.String(allow_none=True, validate=Length(0, 250))
     template_params = fields.String(allow_none=True)
     uuid = fields.UUID(allow_none=True)
 
@@ -192,6 +205,8 @@ class DatasetPutSchema(Schema):
     description = fields.String(allow_none=True)
     main_dttm_col = fields.String(allow_none=True)
     currency_code_column = fields.String(allow_none=True, validate=Length(0, 250))
+    partition_column = fields.String(allow_none=True, validate=Length(0, 250))
+    partition_mapped_column = fields.String(allow_none=True, validate=Length(0, 250))
     normalize_columns = fields.Boolean(allow_none=True, dump_default=False)
     always_filter_main_dttm = fields.Boolean(load_default=False)
     offset = fields.Integer(allow_none=True)
@@ -336,6 +351,10 @@ class ImportV1ColumnSchema(Schema):
     description = fields.String(allow_none=True)
     python_date_format = fields.String(allow_none=True)
     datetime_format = fields.String(allow_none=True)
+    partition_value_transform = fields.String(allow_none=True)
+    # Bundles predating the field must not claim their transform preserves
+    # ordering, which would silently enable range mirroring on import.
+    partition_transform_is_monotonic = fields.Boolean(load_default=False)
     uuid = fields.UUID(allow_none=True)
 
 
@@ -458,6 +477,8 @@ class ImportV1DatasetSchema(Schema):
     external_url = fields.String(allow_none=True)
     normalize_columns = fields.Boolean(load_default=False)
     always_filter_main_dttm = fields.Boolean(load_default=False)
+    partition_column = fields.String(allow_none=True)
+    partition_mapped_column = fields.String(allow_none=True)
     folders = fields.List(fields.Nested(FolderSchema), required=False, allow_none=True)
     # data_file is used by the example loading system to reference Parquet files
     data_file = fields.String(allow_none=True, load_default=None)

superset/mcp_service/common/schema_discovery.py

@@ -160,6 +160,11 @@ _COLUMN_DESCRIPTIONS: dict[str, str] = {
     "filter_select_enabled": "Whether filter select is enabled",
     "normalize_columns": "Whether to normalize column names",
     "always_filter_main_dttm": "Whether to always filter on the main datetime column",
+    "partition_column": "Physical column the engine partitions on",
+    "partition_mapped_column": (
+        "Column whose filters are mirrored onto the partition column; "
+        "defaults to the main datetime column"
+    ),
     "fetch_values_predicate": "SQL predicate for fetching filter values",
     "default_endpoint": "Default endpoint URL",
     "offset": "Row offset for queries",

superset/migrations/versions/2026-09-04_00-00_a7f3c2e91d84_add_partition_filter_mapping.py

@@ -0,0 +1,101 @@
+# 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.
+"""add partition filter mapping
+
+Adds the four columns behind the ``PARTITION_FILTER_MAPPING`` feature:
+
+- ``tables.partition_column`` -- the physical column the engine partitions on
+- ``tables.partition_mapped_column`` -- explicit override for the column whose
+  filters are mirrored; NULL means "follow ``main_dttm_col``"
+- ``table_columns.partition_value_transform`` -- the ``:value`` expression
+- ``table_columns.partition_transform_is_monotonic`` -- gates range operators
+
+The Continuum shadow tables get the same columns so dataset version history and
+restore keep working.
+
+Revision ID: a7f3c2e91d84
+Revises: 1072de5ed955
+Create Date: 2026-08-31 22:30:00.000000
+
+"""
+
+import sqlalchemy as sa
+
+from superset.migrations.shared.utils import add_columns, drop_columns
+
+# revision identifiers, used by Alembic.
+revision = "a7f3c2e91d84"
+down_revision = "7e2c9a4f1b83"
+
+
+def upgrade():
+    add_columns(
+        "tables",
+        sa.Column("partition_column", sa.String(250), nullable=True),
+        sa.Column("partition_mapped_column", sa.String(250), nullable=True),
+    )
+    add_columns(
+        "table_columns",
+        sa.Column("partition_value_transform", sa.Text(), nullable=True),
+        # Nullable, like the other booleans on this table. The legacy
+        # datasource editor saves through `update_from_object`, which writes
+        # NULL for any field its payload omits; NOT NULL here fails that save.
+        # Readers coerce with `bool(...)`, so NULL reads as "not declared" and
+        # ranges stop mirroring rather than mirroring unsoundly.
+        sa.Column(
+            "partition_transform_is_monotonic",
+            sa.Boolean(),
+            nullable=True,
+            server_default=sa.false(),
+        ),
+    )
+
+    # Shadow tables are nullable throughout -- a version row records the state
+    # of the columns that changed, so every column has to tolerate NULL.
+    add_columns(
+        "tables_version",
+        sa.Column("partition_column", sa.String(250), nullable=True),
+        sa.Column("partition_mapped_column", sa.String(250), nullable=True),
+    )
+    add_columns(
+        "table_columns_version",
+        sa.Column("partition_value_transform", sa.Text(), nullable=True),
+        sa.Column("partition_transform_is_monotonic", sa.Boolean(), nullable=True),
+    )
+
+
+def downgrade():
+    drop_columns(
+        "table_columns_version",
+        "partition_value_transform",
+        "partition_transform_is_monotonic",
+    )
+    drop_columns(
+        "tables_version",
+        "partition_column",
+        "partition_mapped_column",
+    )
+    drop_columns(
+        "table_columns",
+        "partition_value_transform",
+        "partition_transform_is_monotonic",
+    )
+    drop_columns(
+        "tables",
+        "partition_column",
+        "partition_mapped_column",
+    )

superset/sql/parse.py

@@ -870,6 +870,44 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         }
     )
 
+    def get_niladic_functions(self) -> set[str]:
+        """
+        Names of functions called with no arguments.
+
+        Some functions mean something entirely different with an empty argument
+        list: on Hive and Impala ``unix_timestamp()`` is the current time while
+        ``unix_timestamp(x)`` is a pure conversion. Callers that care about
+        determinism have to tell those apart by arity, so a name-based check
+        like ``check_functions_present`` is not enough.
+        """
+        niladic: set[str] = set()
+        for function in self._parsed.find_all(exp.Func):
+            sql_name = function.sql_name()
+            name = function.name.upper() if sql_name == "ANONYMOUS" else sql_name
+            if not self._function_args(function):
+                niladic.add(name.upper())
+        return niladic
+
+    @staticmethod
+    def _function_args(function: exp.Func) -> list[Any]:
+        """
+        The arguments a function node was called with.
+
+        `exp.Anonymous` keeps the function *name* in `this` and the arguments in
+        `expressions`, while a named node like `exp.Lower` keeps its single
+        argument in `this` -- so the two shapes have to be read differently or
+        every anonymous call looks like it takes one argument.
+        """
+        if isinstance(function, exp.Anonymous):
+            return list(function.expressions or [])
+
+        args: list[Any] = []
+        for value in function.args.values():
+            if value is None:
+                continue
+            args.extend(value if isinstance(value, list) else [value])
+        return args
+
     def __init__(
         self,
         statement: str | None = None,

superset/superset_typing.py

@@ -292,6 +292,9 @@ class ExplorableData(TypedDict, total=False):
         time_grain_sqla: Available time grains
         main_dttm_col: Main datetime column
         currency_code_column: Column containing currency codes for dynamic formatting
+        partition_column: Physical column the engine partitions on
+        partition_mapped_column: Explicit override for the mirrored column
+        partition_filter_mapping: Summary of the active mapping, or None
         fetch_values_predicate: Predicate for fetching filter values
         template_params: Template parameters for Jinja
         is_sqllab_view: Whether this is a SQL Lab view
@@ -345,6 +348,12 @@ class ExplorableData(TypedDict, total=False):
     extra: str | None
     always_filter_main_dttm: bool
     normalize_columns: bool
+    partition_column: str | None
+    partition_mapped_column: str | None
+    # Self-contained summary for the Explore indicator. Kept separate from
+    # `columns` because `data_for_slices` prunes columns no chart references,
+    # and the partition column is typically referenced by none of them.
+    partition_filter_mapping: dict[str, Any] | None
     rls_filters: list[dict[str, Any]]
     # Set by datasources that cannot return raw row samples (e.g. semantic
     # views, which only expose pre-defined metrics and dimensions).

tests/integration_tests/datasets/commands_tests.py

@@ -118,6 +118,8 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "groupby": True,
                     "is_active": True,
                     "is_dttm": False,
+                    "partition_transform_is_monotonic": False,
+                    "partition_value_transform": None,
                     "python_date_format": None,
                     "type": type_map["source"],
                     "advanced_data_type": None,
@@ -134,6 +136,8 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "groupby": True,
                     "is_active": True,
                     "is_dttm": False,
+                    "partition_transform_is_monotonic": False,
+                    "partition_value_transform": None,
                     "python_date_format": None,
                     "type": type_map["target"],
                     "uuid": column_uuid_map["target"],
@@ -150,6 +154,8 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "groupby": True,
                     "is_active": True,
                     "is_dttm": False,
+                    "partition_transform_is_monotonic": False,
+                    "partition_value_transform": None,
                     "python_date_format": None,
                     "type": type_map["value"],
                     "advanced_data_type": None,
@@ -194,6 +200,8 @@ class TestExportDatasetsCommand(SupersetTestCase):
             "folders": None,
             "normalize_columns": False,
             "always_filter_main_dttm": False,
+            "partition_column": None,
+            "partition_mapped_column": None,
             "offset": 0,
             "params": None,
             "schema": get_example_default_schema(),
@@ -260,6 +268,8 @@ class TestExportDatasetsCommand(SupersetTestCase):
             "normalize_columns",
             "always_filter_main_dttm",
             "folders",
+            "partition_column",
+            "partition_mapped_column",
             "uuid",
             "metrics",
             "columns",

tests/unit_tests/commands/dataset/update_test.py

@@ -92,6 +92,7 @@ def test_update_dataset_sql_authorized_schema(mocker: MockerFixture) -> None:
     mock_dataset.schema = "public"
     mock_dataset.table_name = "test_table"
     mock_dataset.editors = []  # No editors to avoid computation issues
+    mock_dataset.partition_column = None  # No partition filter mapping
 
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_database
@@ -137,6 +138,7 @@ def test_update_dataset_sql_unauthorized_schema(mocker: MockerFixture) -> None:
     mock_dataset.schema = "public"
     mock_dataset.table_name = "test_table"
     mock_dataset.editors = []  # No editors to avoid computation issues
+    mock_dataset.partition_column = None  # No partition filter mapping
 
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_database
@@ -200,6 +202,7 @@ def test_update_dataset_database_id_change_checks_new_database_access(
     mock_dataset.schema = "public"
     mock_dataset.table_name = "test_table"
     mock_dataset.editors = []  # No editors to avoid computation issues
+    mock_dataset.partition_column = None  # No partition filter mapping
 
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_new_database
@@ -256,6 +259,7 @@ def test_update_dataset_database_id_change_allowed_with_access(
     mock_dataset.schema = "public"
     mock_dataset.table_name = "test_table"
     mock_dataset.editors = []  # No editors to avoid computation issues
+    mock_dataset.partition_column = None  # No partition filter mapping
 
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_new_database
@@ -305,6 +309,7 @@ def test_update_dataset_physical_repoint_requires_table_access(
     mock_dataset.table_name = "allowed_table"
     mock_dataset.sql = None  # physical dataset
     mock_dataset.editors = []
+    mock_dataset.partition_column = None
 
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.validate_update_uniqueness.return_value = True
@@ -454,6 +459,7 @@ def test_update_dataset_rejects_malicious_expression(
     mock_dataset.database = mock_database
     mock_dataset.catalog = "catalog"
     mock_dataset.schema = None
+    mock_dataset.partition_column = None  # No partition filter mapping
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_database
     mock_dataset_dao.validate_update_uniqueness.return_value = True
@@ -503,6 +509,7 @@ def test_update_dataset_accepts_benign_expression(mocker: MockerFixture) -> None
     mock_dataset.database = mock_database
     mock_dataset.catalog = "catalog"
     mock_dataset.schema = None
+    mock_dataset.partition_column = None  # No partition filter mapping
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_database
     mock_dataset_dao.validate_update_uniqueness.return_value = True
@@ -544,6 +551,7 @@ def test_update_dataset_accepts_jinja_expression(mocker: MockerFixture) -> None:
     mock_dataset.database = mock_database
     mock_dataset.catalog = "catalog"
     mock_dataset.schema = None
+    mock_dataset.partition_column = None  # No partition filter mapping
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_database
     mock_dataset_dao.validate_update_uniqueness.return_value = True
@@ -1305,6 +1313,7 @@ def test_update_dataset_rejects_malicious_fetch_values_predicate(
     mock_dataset.database = mock_database
     mock_dataset.catalog = "catalog"
     mock_dataset.schema = None
+    mock_dataset.partition_column = None  # No partition filter mapping
     mock_dataset_dao.find_by_id.return_value = mock_dataset
     mock_dataset_dao.get_database_by_id.return_value = mock_database
     mock_dataset_dao.validate_update_uniqueness.return_value = True

tests/unit_tests/connectors/sqla/partition_mapping_test.py

@@ -0,0 +1,234 @@
+# 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.
+"""Unit tests for ``superset.connectors.sqla.partition_mapping``."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from superset.connectors.sqla.partition_mapping import (
+    contains_jinja,
+    contains_value_placeholder,
+    find_non_deterministic_functions,
+    MappingValidationIssue,
+    validate_partition_mapping,
+)
+
+# ---------------------------------------------------------------------------
+# §5 — transform inspection helpers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "transform,expected",
+    [
+        ("unix_timestamp(:value)", True),
+        ("lower(:value)", True),
+        ("CAST(:value AS BIGINT)", True),
+        ("unix_timestamp(event_time)", False),
+        (":values", False),
+        ("", False),
+        (None, False),
+    ],
+)
+def test_contains_value_placeholder(transform: str | None, expected: bool) -> None:
+    assert contains_value_placeholder(transform) is expected
+
+
+@pytest.mark.parametrize(
+    "transform,expected",
+    [
+        ("unix_timestamp(:value)", False),
+        ("{{ current_username() }}", True),
+        ("lower({% if x %}:value{% endif %})", True),
+        ("lower(:value) -- {# comment #}", True),
+    ],
+)
+def test_contains_jinja(transform: str, expected: bool) -> None:
+    assert contains_jinja(transform) is expected
+
+
+@pytest.mark.parametrize(
+    "transform",
+    [
+        "unix_timestamp(:value)",
+        "lower(:value)",
+        "CAST(:value AS BIGINT)",
+        "date_format(:value, 'yyyyMMdd')",
+    ],
+)
+def test_pure_transforms_report_no_non_deterministic_functions(
+    transform: str,
+) -> None:
+    assert find_non_deterministic_functions(transform, "hive") == set()
+
+
+@pytest.mark.parametrize(
+    "transform,expected_name",
+    [
+        ("date_diff(:value, now())", "NOW"),
+        ("CAST(:value AS BIGINT) + rand()", "RAND"),
+        ("CAST(:value AS DATE) - current_date", "CURRENT_DATE"),
+    ],
+)
+def test_non_deterministic_functions_are_reported(
+    transform: str, expected_name: str
+) -> None:
+    """
+    The probe runs at a different moment and in a different session from the
+    chart query, and its result is cached, so anything time- or
+    randomness-dependent freezes a snapshot of probe time into the predicate.
+    """
+    assert expected_name in find_non_deterministic_functions(transform, "hive")
+
+
+def test_niladic_unix_timestamp_is_rejected_but_the_unary_form_is_not() -> None:
+    """
+    On Hive/Impala ``unix_timestamp()`` means "now" while ``unix_timestamp(x)``
+    -- the canonical temporal transform -- is pure. The distinction is the whole
+    reason this check inspects arity rather than just the name.
+    """
+    assert find_non_deterministic_functions("unix_timestamp(:value)", "hive") == set()
+    assert find_non_deterministic_functions(
+        "unix_timestamp(:value) - unix_timestamp()", "hive"
+    )
+
+
+# ---------------------------------------------------------------------------
+# §5 — save-time validation, in two tiers
+# ---------------------------------------------------------------------------
+
+
+def _issues(**kwargs: Any) -> list[MappingValidationIssue]:
+    defaults: dict[str, Any] = {
+        "column_names": {"event_time", "dt_epoch", "country", "region_key"},
+        "partition_column": "dt_epoch",
+        "partition_mapped_column": None,
+        "main_dttm_col": "event_time",
+        "transform": "unix_timestamp(:value)",
+        "engine": "hive",
+    }
+    defaults.update(kwargs)
+    return validate_partition_mapping(**defaults)
+
+
+def _blocking(issues: list[MappingValidationIssue]) -> list[MappingValidationIssue]:
+    return [issue for issue in issues if issue.blocking]
+
+
+def _warnings(issues: list[MappingValidationIssue]) -> list[MappingValidationIssue]:
+    return [issue for issue in issues if not issue.blocking]
+
+
+def test_a_well_formed_mapping_raises_nothing() -> None:
+    assert _issues() == []
+
+
+def test_no_partition_column_means_nothing_to_validate() -> None:
+    assert _issues(partition_column=None, transform=None) == []
+
+
+# Tier 1 — blocks the save
+
+
+def test_an_unknown_partition_column_blocks_the_save() -> None:
+    issues = _issues(partition_column="nope")
+    assert len(_blocking(issues)) == 1
+    assert issues[0].field == "partition_column"
+
+
+def test_an_unknown_mapped_column_override_blocks_the_save() -> None:
+    issues = _issues(partition_mapped_column="nope")
+    assert len(_blocking(issues)) == 1
+    assert issues[0].field == "partition_mapped_column"
+
+
+def test_an_explicit_self_mapping_blocks_the_save() -> None:
+    issues = _blocking(_issues(partition_mapped_column="dt_epoch"))
+    assert len(issues) == 1
+    assert "itself" in issues[0].message
+
+
+def test_an_implicit_self_mapping_blocks_the_save() -> None:
+    """
+    Checking only the explicit override misses the case an owner actually hits:
+    setting ``partition_column`` to the column that is *already* the default
+    datetime column, with no override in play.
+    """
+    issues = _blocking(
+        _issues(partition_column="event_time", main_dttm_col="event_time")
+    )
+    assert len(issues) == 1
+    assert "itself" in issues[0].message
+
+
+def test_jinja_in_the_transform_blocks_the_save() -> None:
+    """
+    The probe would render the template in a different context at a different
+    time from the chart query, so v1 disallows it outright.
+    """
+    issues = _blocking(_issues(transform="unix_timestamp('{{ ds }}' , :value)"))
+    assert len(issues) == 1
+    assert "Jinja" in issues[0].message
+
+
+@pytest.mark.parametrize(
+    "transform",
+    [
+        "unix_timestamp(:value) - unix_timestamp()",
+        "date_diff(:value, now())",
+        "CAST(:value AS BIGINT) + rand()",
+    ],
+)
+def test_a_non_deterministic_transform_blocks_the_save(transform: str) -> None:
+    issues = _blocking(_issues(transform=transform))
+    assert len(issues) == 1
+    assert issues[0].field == "partition_value_transform"
+
+
+# Tier 2 — saves, but the mapping stays inactive
+
+
+def test_an_unparseable_transform_saves_with_a_warning() -> None:
+    """The PRD is explicit: a bad transform still saves, it just stays inactive."""
+    issues = _issues(transform="unix_timestamp(:value")
+    assert _blocking(issues) == []
+    assert len(_warnings(issues)) == 1
+
+
+def test_a_transform_without_the_placeholder_saves_with_a_warning() -> None:
+    issues = _issues(transform="unix_timestamp(event_time)")
+    assert _blocking(issues) == []
+    assert len(_warnings(issues)) == 1
+
+
+def test_a_missing_transform_saves_with_a_warning() -> None:
+    issues = _issues(transform=None)
+    assert _blocking(issues) == []
+    assert len(_warnings(issues)) == 1
+
+
+def test_an_unparseable_transform_skips_the_checks_that_need_a_parse() -> None:
+    """
+    The Jinja and non-determinism checks require a successful parse. When there
+    is nothing to inspect, fall through to a warning rather than reporting a
+    blocking error the owner cannot act on.
+    """
+    issues = _issues(transform="now(:value")
+    assert _blocking(issues) == []

tests/unit_tests/datasets/commands/export_test.py

@@ -216,6 +216,8 @@ folders:
     - uuid: 00000000-0000-0000-0000-000000000005
       type: column
       name: profit
+partition_column: null
+partition_mapped_column: null
 uuid: {payload["uuid"]}
 metrics:
 - metric_name: cnt
@@ -244,6 +246,8 @@ columns:
   datetime_format: null
   extra:
     certified_by: User
+  partition_value_transform: null
+  partition_transform_is_monotonic: false
   uuid: 00000000-0000-0000-0000-000000000005
 - column_name: ds
   verbose_name: null
@@ -258,6 +262,8 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
+  partition_value_transform: null
+  partition_transform_is_monotonic: false
   uuid: 00000000-0000-0000-0000-000000000006
 - column_name: user_id
   verbose_name: null
@@ -272,6 +278,8 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
+  partition_value_transform: null
+  partition_transform_is_monotonic: false
   uuid: 00000000-0000-0000-0000-000000000007
 - column_name: revenue
   verbose_name: null
@@ -286,6 +294,8 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
+  partition_value_transform: null
+  partition_transform_is_monotonic: false
   uuid: 00000000-0000-0000-0000-000000000008
 - column_name: expenses
   verbose_name: null
@@ -300,6 +310,8 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
+  partition_value_transform: null
+  partition_transform_is_monotonic: false
   uuid: 00000000-0000-0000-0000-000000000009
 version: 1.0.0
 database_uuid: {database.uuid}

tests/unit_tests/datasets/partition_mapping_serialization_test.py

@@ -0,0 +1,266 @@
+# 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.
+"""
+The partition mapping has to survive every layer it passes through.
+
+``always_filter_main_dttm`` is the template these follow: a dataset-level
+setting that names a column and appears in the ORM, the export fields, the
+``data`` payload, the API schemas and the frontend types. A field missing from
+any one of them is dropped silently, which is exactly the failure mode these
+tests exist to catch.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from flask import Flask
+
+from superset.connectors.sqla.models import SqlaTable, TableColumn
+from superset.datasets.schemas import (
+    DatasetColumnsPutSchema,
+    DatasetPutSchema,
+    ImportV1ColumnSchema,
+    ImportV1DatasetSchema,
+)
+from superset.models.core import Database
+
+DATASET_FIELDS = ["partition_column", "partition_mapped_column"]
+COLUMN_FIELDS = ["partition_value_transform", "partition_transform_is_monotonic"]
+
+
+@pytest.fixture(autouse=True)
+def enable_partition_filter_mapping(app: Flask) -> Any:
+    app.config["DEFAULT_FEATURE_FLAGS"]["PARTITION_FILTER_MAPPING"] = True
+    yield
+    del app.config["DEFAULT_FEATURE_FLAGS"]["PARTITION_FILTER_MAPPING"]
+
+
+def _table() -> SqlaTable:
+    database = Database(database_name="test_db", sqlalchemy_uri="sqlite://")
+    column = TableColumn(column_name="event_time", is_dttm=True, type="TIMESTAMP")
+    column.partition_value_transform = "unix_timestamp(:value)"
+    column.partition_transform_is_monotonic = True
+    table = SqlaTable(
+        table_name="web_events",
+        database=database,
+        main_dttm_col="event_time",
+        columns=[column, TableColumn(column_name="dt_epoch", type="BIGINT")],
+    )
+    table.partition_column = "dt_epoch"
+    table.partition_mapped_column = None
+    return table
+
+
+@pytest.mark.parametrize("field", DATASET_FIELDS)
+def test_dataset_fields_are_exported(field: str) -> None:
+    """
+    Being in ``export_fields`` is what makes the mapping travel in dataset YAML,
+    and what makes ``update_from_object`` write it back on import.
+    """
+    assert field in SqlaTable.export_fields
+
+
+@pytest.mark.parametrize("field", COLUMN_FIELDS)
+def test_column_fields_are_exported(field: str) -> None:
+    assert field in TableColumn.export_fields
+
+
+@pytest.mark.parametrize("field", DATASET_FIELDS)
+def test_dataset_fields_reach_the_explore_payload(app: Flask, field: str) -> None:
+    with app.app_context():
+        data = _table().data
+    assert field in data
+
+
+@pytest.mark.parametrize("field", COLUMN_FIELDS)
+def test_column_fields_reach_the_explore_payload(app: Flask, field: str) -> None:
+    with app.app_context():
+        data = _table().data
+    assert field in data["columns"][0]
+
+
+def test_the_mapping_summary_survives_dashboard_payload_pruning(app: Flask) -> None:
+    """
+    ``data_for_slices`` prunes columns no chart references, and the partition
+    column is typically referenced by none of them. The Explore indicator
+    therefore reads a self-contained dataset-level dict rather than looking the
+    column up inside ``datasource.columns``.
+    """
+    with app.app_context():
+        data = _table().data_for_slices([])
+
+    assert data["partition_filter_mapping"] == {
+        "partition_column": "dt_epoch",
+        "mapped_column": "event_time",
+        "active": True,
+    }
+
+
+def test_the_mapping_summary_reports_inactive_without_a_transform(
+    app: Flask,
+) -> None:
+    table = _table()
+    table.columns[0].partition_value_transform = None
+
+    with app.app_context():
+        summary = table.data["partition_filter_mapping"]
+
+    assert summary is not None
+    assert summary["active"] is False
+
+
+def test_there_is_no_mapping_summary_without_a_partition_column(
+    app: Flask,
+) -> None:
+    table = _table()
+    table.partition_column = None
+
+    with app.app_context():
+        assert table.data["partition_filter_mapping"] is None
+
+
+@pytest.mark.parametrize("field", DATASET_FIELDS)
+def test_put_schema_accepts_the_dataset_fields(field: str) -> None:
+    loaded = DatasetPutSchema().load({field: "dt_epoch"})
+    assert loaded[field] == "dt_epoch"
+
+
+def test_put_schema_accepts_the_column_fields() -> None:
+    loaded = DatasetColumnsPutSchema().load(
+        {
+            "column_name": "event_time",
+            "partition_value_transform": "unix_timestamp(:value)",
+            "partition_transform_is_monotonic": True,
+        }
+    )
+    assert loaded["partition_value_transform"] == "unix_timestamp(:value)"
+    assert loaded["partition_transform_is_monotonic"] is True
+
+
+def test_put_schema_allows_clearing_the_mapping() -> None:
+    """Removing a mapping is a null, not an omission."""
+    loaded = DatasetPutSchema().load({"partition_column": None})
+    assert loaded["partition_column"] is None
+
+
+def test_import_schema_round_trips_the_mapping() -> None:
+    loaded = ImportV1DatasetSchema().load(
+        {
+            "table_name": "web_events",
+            "uuid": "00000000-0000-0000-0000-000000000001",
+            "database_uuid": "00000000-0000-0000-0000-000000000002",
+            "version": "1.0.0",
+            "partition_column": "dt_epoch",
+            "partition_mapped_column": "event_time",
+        }
+    )
+    assert loaded["partition_column"] == "dt_epoch"
+    assert loaded["partition_mapped_column"] == "event_time"
+
+
+def test_import_column_schema_round_trips_the_transform() -> None:
+    loaded = ImportV1ColumnSchema().load(
+        {
+            "column_name": "event_time",
+            "partition_value_transform": "unix_timestamp(:value)",
+            "partition_transform_is_monotonic": True,
+        }
+    )
+    assert loaded["partition_value_transform"] == "unix_timestamp(:value)"
+    assert loaded["partition_transform_is_monotonic"] is True
+
+
+def test_import_column_schema_defaults_the_monotonic_flag_to_false() -> None:
+    """
+    The flag gates range mirroring. A dataset imported from a bundle that
+    predates the field must not silently claim its transform preserves ordering.
+    """
+    loaded = ImportV1ColumnSchema().load({"column_name": "event_time"})
+    assert loaded["partition_transform_is_monotonic"] is False
+
+
+@pytest.mark.parametrize("field", DATASET_FIELDS)
+def test_the_api_exposes_and_accepts_the_dataset_fields(field: str) -> None:
+    from superset.datasets.api import DatasetRestApi
+
+    assert field in DatasetRestApi.show_select_columns
+    assert field in DatasetRestApi.edit_columns
+
+
+# ---------------------------------------------------------------------------
+# §10 — a column sync can pull the partition column out from under the mapping
+# ---------------------------------------------------------------------------
+
+
+def test_a_sync_that_removes_the_partition_column_clears_the_mapping() -> None:
+    """
+    An API-driven ``override_columns=true`` sync must not leave a dangling
+    mapping. This is the authoritative path -- the client-side sync clears the
+    mapping too, but a caller can bypass the editor entirely.
+    """
+    from superset.daos.dataset import DatasetDAO
+
+    table = _table()
+    DatasetDAO.clear_dangling_partition_mapping(table, {"event_time"})
+
+    assert table.partition_column is None
+    assert table.partition_mapped_column is None
+
+
+def test_a_sync_that_removes_the_mapped_column_clears_only_the_override() -> None:
+    """
+    The partition column is still real, so the designation survives; the mapping
+    falls back to "no mapped column" and goes inactive until one is chosen.
+    """
+    from superset.daos.dataset import DatasetDAO
+
+    table = _table()
+    table.partition_mapped_column = "event_time"
+
+    DatasetDAO.clear_dangling_partition_mapping(table, {"dt_epoch"})
+
+    assert table.partition_column == "dt_epoch"
+    assert table.partition_mapped_column is None
+
+
+def test_a_sync_that_keeps_both_columns_leaves_the_mapping_alone() -> None:
+    from superset.daos.dataset import DatasetDAO
+
+    table = _table()
+    DatasetDAO.clear_dangling_partition_mapping(table, {"event_time", "dt_epoch"})
+
+    assert table.partition_column == "dt_epoch"
+
+
+@pytest.mark.parametrize("field", COLUMN_FIELDS)
+def test_column_fields_tolerate_a_null_write(field: str) -> None:
+    """
+    The legacy datasource editor saves through `update_from_object`, which does
+    `setattr(self, attr, obj.get(attr))` for every field in
+    `update_from_object_fields` -- so any field its payload omits is written as
+    NULL. A NOT NULL column here makes that save fail with an IntegrityError
+    (surfacing as a 422), which is how this was found.
+    """
+    column = TableColumn.__table__.columns[field]
+    assert column.nullable, f"{field} must be nullable for the legacy save path"
+
+
+@pytest.mark.parametrize("field", DATASET_FIELDS)
+def test_dataset_fields_tolerate_a_null_write(field: str) -> None:
+    assert SqlaTable.__table__.columns[field].nullable

tests/unit_tests/sql/parse_tests.py

@@ -6393,3 +6393,38 @@ def test_has_aggregate(expression: str, expected: bool) -> None:
     function sqlglot can't model.
     """
     assert has_aggregate(expression) is expected
+
+
+@pytest.mark.parametrize(
+    "sql, engine, expected",
+    [
+        # Hive's parser resolves the zero-argument form to CURRENT_TIMESTAMP,
+        # which is what it actually means, so that is the name reported.
+        ("SELECT unix_timestamp()", "hive", {"CURRENT_TIMESTAMP"}),
+        ("SELECT unix_timestamp( )", "hive", {"CURRENT_TIMESTAMP"}),
+        ("SELECT unix_timestamp(ds) - unix_timestamp()", "hive", {"CURRENT_TIMESTAMP"}),
+        # Dialects that do not special-case it report the name as written.
+        ("SELECT unix_timestamp()", "sqlite", {"UNIX_TIMESTAMP"}),
+        ("SELECT unix_timestamp(ds)", "hive", set()),
+        ("SELECT unix_timestamp(ds)", "sqlite", set()),
+        ("SELECT lower(country)", "hive", set()),
+        ("SELECT * FROM some_table", "hive", set()),
+        # A named node whose arguments span all three shapes sqlglot uses: a
+        # scalar (`this`), a list (`expressions`), and unset optional slots
+        # left as `None`. Reading only `this` would count this as niladic.
+        ("SELECT coalesce(a, b)", "hive", set()),
+        # `expressions` is the only argument here, so the list branch is what
+        # decides whether the call looks niladic at all.
+        ("SELECT concat(a, b)", "hive", set()),
+    ],
+)
+def test_get_niladic_functions(sql: str, engine: str, expected: set[str]) -> None:
+    """
+    Check the `get_niladic_functions` method.
+
+    Some functions mean something entirely different with no arguments -- on
+    Hive and Impala `unix_timestamp()` is the current time while
+    `unix_timestamp(x)` is a pure conversion -- so callers that care about
+    determinism need to distinguish the two by arity, not by name.
+    """
+    assert SQLStatement(sql, engine).get_niladic_functions() == expected