You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by ru...@apache.org on 2026/06/23 17:13:30 UTC
(superset) branch master updated: fix(trino): ignore Iceberg $partitions metadata fields in partition detection (#41055)
This is an automated email from the ASF dual-hosted git repository.
rusackas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new ca3c4204124 fix(trino): ignore Iceberg $partitions metadata fields in partition detection (#41055)
ca3c4204124 is described below
commit ca3c4204124f7ea03f296007f50e6e2c3af55073
Author: Alexis <51...@users.noreply.github.com>
AuthorDate: Tue Jun 23 19:13:23 2026 +0200
fix(trino): ignore Iceberg $partitions metadata fields in partition detection (#41055)
Co-authored-by: Evan Rusackas <ev...@preset.io>
---
superset/db_engine_specs/presto.py | 11 +-
superset/db_engine_specs/trino.py | 122 +++++++++++++++++++-
tests/unit_tests/db_engine_specs/test_trino.py | 150 +++++++++++++++++++++++++
3 files changed, 277 insertions(+), 6 deletions(-)
diff --git a/superset/db_engine_specs/presto.py b/superset/db_engine_specs/presto.py
index 4f4d2eb19f5..b9ca92c75e5 100644
--- a/superset/db_engine_specs/presto.py
+++ b/superset/db_engine_specs/presto.py
@@ -636,6 +636,7 @@ class PrestoBaseEngineSpec(BaseEngineSpec, metaclass=ABCMeta):
cls,
database: Database,
table: Table,
+ indexes: list[dict[str, Any]] | None = None,
**kwargs: Any,
) -> Any:
"""Returns the latest (max) partition value for a table
@@ -660,7 +661,15 @@ class PrestoBaseEngineSpec(BaseEngineSpec, metaclass=ABCMeta):
>>> latest_sub_partition('sub_partition_table', event_type='click')
'2018-01-01'
"""
- indexes = database.get_indexes(table)
+ if indexes is None:
+ indexes = database.get_indexes(table)
+
+ if not indexes:
+ raise SupersetTemplateException(
+ f"Error getting partition for {table}. "
+ "Verify that this table has a partition."
+ )
+
part_fields = indexes[0]["column_names"]
for k in kwargs.keys(): # pylint: disable=consider-iterating-dictionary
if k not in k in part_fields: # pylint: disable=comparison-with-itself
diff --git a/superset/db_engine_specs/trino.py b/superset/db_engine_specs/trino.py
index 6bacfc95355..0747a8ce4a2 100644
--- a/superset/db_engine_specs/trino.py
+++ b/superset/db_engine_specs/trino.py
@@ -21,6 +21,7 @@ import logging
import math
import threading
import time
+from collections.abc import Sequence
from typing import Any, TYPE_CHECKING
import requests
@@ -30,7 +31,7 @@ from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.engine.url import URL
from sqlalchemy.exc import NoSuchTableError
-from superset import db
+from superset import cache_manager, db
from superset.common.db_query_status import QueryStatus
from superset.constants import QUERY_CANCEL_KEY, QUERY_EARLY_CANCEL_KEY
from superset.db_engine_specs.base import (
@@ -71,6 +72,18 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
engine_name = "Trino"
allows_alias_to_source_column = False
+ # The full set of columns Trino's "<table>$partitions" exposes for an
+ # Iceberg table. The real partition keys are nested in the "partition" ROW,
+ # so none of these are user partition columns.
+ iceberg_partitions_metadata_columns = frozenset(
+ {"partition", "record_count", "file_count", "total_size", "data"}
+ )
+ # Always present for Iceberg; used as the positive signal so we don't act on
+ # a table that merely happens to share one of the names above.
+ iceberg_partitions_signature_columns = frozenset(
+ {"record_count", "file_count", "total_size"}
+ )
+
metadata = {
"description": (
"Trino is a distributed SQL query engine for big data analytics."
@@ -166,6 +179,37 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
and "error 401" in str(ex)
)
+ @classmethod
+ def _filter_iceberg_partition_indexes(
+ cls,
+ indexes: list[dict[str, Any]] | None,
+ ) -> list[dict[str, Any]]:
+ """
+ Drop Iceberg "$partitions" metadata indexes.
+
+ A partition index is recognized as Iceberg metadata only when it carries
+ the signature columns *and* every one of its columns is a known metadata
+ field. Requiring the latter means an index with any real partition key
+ (e.g. a Hive table partitioned on "ds" that also has a column named
+ "record_count") is left untouched. Such an index has no real partition
+ keys, so it's dropped entirely; all other indexes pass through unchanged.
+
+ :param indexes: the indexes associated with a table
+ :returns: the indexes with Iceberg metadata indexes removed
+ """
+ filtered_indexes = []
+ for index in indexes or []:
+ column_names = set(index.get("column_names") or [])
+ is_iceberg_metadata = (
+ index.get("name") == "partition"
+ and cls.iceberg_partitions_signature_columns <= column_names
+ and column_names <= cls.iceberg_partitions_metadata_columns
+ )
+ if not is_iceberg_metadata:
+ filtered_indexes.append(index)
+
+ return filtered_indexes
+
@classmethod
def get_extra_table_metadata(
cls,
@@ -174,7 +218,9 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
) -> dict[str, Any]:
metadata = {}
- if indexes := database.get_indexes(table):
+ if indexes := cls._filter_iceberg_partition_indexes(
+ database.get_indexes(table)
+ ):
col_names, latest_parts = cls.latest_partition(
database,
table,
@@ -182,8 +228,11 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
indexes=indexes,
)
- if not latest_parts:
- latest_parts = tuple([None] * len(col_names))
+ partition_values: Sequence[str | None]
+ if latest_parts:
+ partition_values = latest_parts
+ else:
+ partition_values = [None] * len(col_names)
metadata["partitions"] = {
"cols": sorted( # noqa: C414
@@ -196,7 +245,7 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
}
)
),
- "latest": dict(zip(col_names, latest_parts, strict=False)),
+ "latest": dict(zip(col_names, partition_values, strict=False)),
"partitionQuery": cls._partition_query(
table=table,
indexes=indexes,
@@ -631,3 +680,66 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
return super().get_indexes(database, inspector, table)
except NoSuchTableError:
return []
+
+ @classmethod
+ @cache_manager.data_cache.memoize(timeout=60)
+ def latest_partition(
+ cls,
+ database: Database,
+ table: Table,
+ show_first: bool = False,
+ indexes: list[dict[str, Any]] | None = None,
+ ) -> tuple[list[str], list[str] | None]:
+ """
+ Return the latest partition for a table.
+
+ Iceberg "$partitions" metadata fields are filtered out first, so we
+ never build a latest-partition query against them. Memoized like the
+ base implementation so the index lookup is not repeated on cache hits.
+
+ :param database: the database the query will be run against
+ :param table: the table instance
+ :param show_first: return the value for the first partitioning key when
+ there are several
+ :param indexes: the indexes associated with the table
+ :returns: the column names and the latest partition values
+ """
+ if indexes is None:
+ indexes = database.get_indexes(table)
+
+ return super().latest_partition(
+ database,
+ table,
+ show_first=show_first,
+ indexes=cls._filter_iceberg_partition_indexes(indexes),
+ )
+
+ @classmethod
+ def latest_sub_partition(
+ cls,
+ database: Database,
+ table: Table,
+ indexes: list[dict[str, Any]] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ """
+ Return the latest sub-partition value for a table.
+
+ Iceberg "$partitions" metadata fields are filtered out first, so the
+ ``latest_sub_partition`` macro never builds a query against them.
+
+ :param database: the database the query will be run against
+ :param table: the table instance
+ :param indexes: the indexes associated with the table
+ :param kwargs: filtering criteria on the partition list
+ :returns: the latest sub-partition value
+ """
+ if indexes is None:
+ indexes = database.get_indexes(table)
+
+ return super().latest_sub_partition(
+ database,
+ table,
+ indexes=cls._filter_iceberg_partition_indexes(indexes),
+ **kwargs,
+ )
diff --git a/tests/unit_tests/db_engine_specs/test_trino.py b/tests/unit_tests/db_engine_specs/test_trino.py
index 5f8f058ba2a..fcce320208e 100644
--- a/tests/unit_tests/db_engine_specs/test_trino.py
+++ b/tests/unit_tests/db_engine_specs/test_trino.py
@@ -344,6 +344,156 @@ def test_get_extra_table_metadata(mocker: MockerFixture) -> None:
assert result["partitions"]["latest"] == {"ds": "01-01-19", "hour": 1}
+def test_get_extra_table_metadata_iceberg_unpartitioned(
+ mocker: MockerFixture,
+) -> None:
+ """
+ Iceberg ``$partitions`` metadata fields must not be treated as partition
+ columns, so an unpartitioned Iceberg table yields no partition metadata and
+ no latest-partition query.
+ """
+ from superset.db_engine_specs.trino import TrinoEngineSpec
+
+ db_mock = mocker.MagicMock()
+ db_mock.get_indexes = Mock(
+ return_value=[
+ {
+ "name": "partition",
+ "column_names": ["data", "file_count", "record_count", "total_size"],
+ }
+ ]
+ )
+ db_mock.get_extra = Mock(return_value={})
+ db_mock.has_view = Mock(return_value=None)
+ db_mock.get_df = Mock()
+ result = TrinoEngineSpec.get_extra_table_metadata(
+ db_mock,
+ Table("test_table", "test_schema"),
+ )
+ assert "partitions" not in result
+ db_mock.get_df.assert_not_called()
+
+
+def test_get_extra_table_metadata_iceberg_partitioned(
+ mocker: MockerFixture,
+) -> None:
+ """
+ A genuinely partitioned Iceberg table exposes a ``partition`` ROW column
+ alongside the metadata fields. Since the real partition columns are nested
+ in the ROW and not directly queryable here, the partition block is skipped
+ rather than generating an invalid latest-partition query.
+ """
+ from superset.db_engine_specs.trino import TrinoEngineSpec
+
+ db_mock = mocker.MagicMock()
+ db_mock.get_indexes = Mock(
+ return_value=[
+ {
+ "name": "partition",
+ "column_names": [
+ "partition",
+ "record_count",
+ "file_count",
+ "total_size",
+ "data",
+ ],
+ }
+ ]
+ )
+ db_mock.get_extra = Mock(return_value={})
+ db_mock.has_view = Mock(return_value=None)
+ db_mock.get_df = Mock()
+ result = TrinoEngineSpec.get_extra_table_metadata(
+ db_mock,
+ Table("test_table", "test_schema"),
+ )
+ assert "partitions" not in result
+ db_mock.get_df.assert_not_called()
+
+
+def test_latest_sub_partition_iceberg(mocker: MockerFixture) -> None:
+ """
+ The ``latest_sub_partition`` macro must not query Iceberg ``$partitions``
+ metadata fields: with no real partition keys it raises instead of building
+ SQL.
+ """
+ from superset.db_engine_specs.trino import TrinoEngineSpec
+ from superset.exceptions import SupersetTemplateException
+
+ db_mock = mocker.MagicMock()
+ db_mock.get_indexes = Mock(
+ return_value=[
+ {
+ "name": "partition",
+ "column_names": ["data", "file_count", "record_count", "total_size"],
+ }
+ ]
+ )
+ db_mock.get_df = Mock()
+ with pytest.raises(SupersetTemplateException):
+ TrinoEngineSpec.latest_sub_partition(
+ db_mock,
+ Table("test_table", "test_schema"),
+ record_count="1",
+ )
+ db_mock.get_df.assert_not_called()
+
+
+def test_filter_iceberg_partition_indexes() -> None:
+ """
+ Iceberg ``$partitions`` metadata indexes are dropped while real partition
+ indexes (and any index sharing only some metadata names) pass through.
+ """
+ from superset.db_engine_specs.trino import TrinoEngineSpec
+
+ # Iceberg metadata-only partition index is dropped entirely.
+ assert (
+ TrinoEngineSpec._filter_iceberg_partition_indexes(
+ [
+ {
+ "name": "partition",
+ "column_names": [
+ "data",
+ "file_count",
+ "record_count",
+ "total_size",
+ ],
+ }
+ ]
+ )
+ == []
+ )
+
+ # Hive partition index (no Iceberg signature) is returned unchanged.
+ hive_indexes = [{"name": "partition", "column_names": ["ds", "hour"]}]
+ assert (
+ TrinoEngineSpec._filter_iceberg_partition_indexes(hive_indexes) == hive_indexes
+ )
+
+ # A Hive partition column that happens to be named ``data`` is preserved
+ # because the Iceberg signature columns are absent.
+ data_indexes = [{"name": "partition", "column_names": ["data"]}]
+ assert (
+ TrinoEngineSpec._filter_iceberg_partition_indexes(data_indexes) == data_indexes
+ )
+
+ # A real partition key alongside coincidental signature-named columns is
+ # preserved untouched: not every column is an Iceberg metadata field.
+ mixed_indexes = [
+ {
+ "name": "partition",
+ "column_names": ["ds", "record_count", "file_count", "total_size"],
+ }
+ ]
+ assert (
+ TrinoEngineSpec._filter_iceberg_partition_indexes(mixed_indexes)
+ == mixed_indexes
+ )
+
+ # Empty / falsy input yields an empty list.
+ assert TrinoEngineSpec._filter_iceberg_partition_indexes(None) == []
+
+
@patch("sqlalchemy.engine.Engine.connect")
def test_cancel_query_success(engine_mock: Mock) -> None:
from superset.db_engine_specs.trino import TrinoEngineSpec