Compare commits

...
Author SHA1 Message Date
Hugh A. Miles IIandClaude Opus 5 f21009f6b8 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>
2026-09-05 15:19:30 -03:00
20 changed files with 1337 additions and 0 deletions
+6
View File
@@ -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,
+53
View File
@@ -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,
+103
View File
@@ -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,
+4
View File
@@ -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
+67
View File
@@ -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:
@@ -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 []
+33
View File
@@ -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]]
+3
View File
@@ -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()
+4
View File
@@ -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",
+21
View File
@@ -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)
@@ -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",
@@ -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",
)
+38
View File
@@ -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,
+9
View File
@@ -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).
@@ -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",
@@ -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
@@ -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) == []
@@ -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}
@@ -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
+35
View File
@@ -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