sadpandajoe commented on code in PR #44849: URL: https://github.com/apache/superset/pull/44849#discussion_r4209416669
########## superset/semantic_layers/metadata_binding.py: ########## @@ -0,0 +1,308 @@ +# 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. + +"""Host construction and operation lifetime for optional shared metadata.""" + +from __future__ import annotations + +import logging +import math +import time +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar, Token +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any, TYPE_CHECKING + +from flask import current_app, has_app_context, has_request_context, request +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session +from superset_core.semantic_layers.layer import SemanticLayer as LayerABC +from superset_core.semantic_layers.metadata import ( + CatalogSnapshot, + MetadataRefreshError, + remaining_budget, +) +from superset_core.semantic_layers.view import SemanticView as ViewABC + +from superset import db, is_feature_enabled +from superset.coordination.deadline_backend import DeadlineRedisBackend +from superset.semantic_layers.metadata import ( + FETCH_DEADLINE_SECONDS, + metadata_scope, + ScopedMetadataStore, +) +from superset.semantic_layers.registry import registry +from superset.utils import json + +if TYPE_CHECKING: + from superset.semantic_layers.models import SemanticLayer, SemanticView + + +logger: logging.Logger = logging.getLogger(__name__) + + +@dataclass +class MetadataOperation: + """One request or worker operation; nested discovery shares its deadline.""" + + deadline: float + layers: dict[str, LayerABC[Any, ViewABC]] = field(default_factory=dict) + views: dict[tuple[str, str, str], ViewABC] = field(default_factory=dict) + stores: dict[str, ScopedMetadataStore] = field(default_factory=dict) + configurations: dict[str, dict[str, Any]] = field(default_factory=dict) + + +_OPERATION_KEY: str = "superset.semantic_metadata.operation" +_worker_operation: ContextVar[MetadataOperation | None] = ContextVar( + _OPERATION_KEY, default=None +) + +_worker_chart: ContextVar[bool] = ContextVar( + "superset.semantic_metadata.chart", default=False +) + + +def request_metadata_budget() -> None: + """Register before authentication hooks; this performs no provider or cache I/O.""" + if current_app.config.get("SEMANTIC_LAYER_METADATA_REFRESH_ENABLED") is True: + request.environ.setdefault( + _OPERATION_KEY, MetadataOperation(time.monotonic() + FETCH_DEADLINE_SECONDS) + ) + + +def _current_operation() -> MetadataOperation | None: + if has_request_context(): + return request.environ.get(_OPERATION_KEY) or _worker_operation.get() + return _worker_operation.get() + + +def _operation(*, require_budget: bool = True) -> MetadataOperation: + state: MetadataOperation | None = _current_operation() + if state is None or not math.isfinite(state.deadline): + raise MetadataRefreshError("configuration") + if require_budget: + remaining_budget(state.deadline, now=time.monotonic()) + return state + + +def operation_deadline() -> float: + return _operation().deadline + + +@contextmanager +def metadata_operation(*, deadline: float | None = None) -> Iterator[None]: + """Workers opt in before access checks; nested calls never replenish the budget.""" + if deadline is not None and not math.isfinite(deadline): + raise MetadataRefreshError("configuration") + if deadline is not None: + remaining_budget(deadline, now=time.monotonic()) + if _current_operation() is not None: + _operation(require_budget=False) + yield + return + if has_request_context(): + # HTTP requests must enter through the registered early request hook. + raise MetadataRefreshError("configuration") + ceiling: float = time.monotonic() + FETCH_DEADLINE_SECONDS + state: MetadataOperation = MetadataOperation( + min(deadline, ceiling) if deadline is not None else ceiling + ) + token: Token[MetadataOperation | None] = _worker_operation.set(state) + try: + operation_deadline() + yield + finally: + _worker_operation.reset(token) + + +@contextmanager +def chart_metadata_operation() -> Iterator[None]: + """Give each worker chart a fresh budget; nested chart work shares it.""" + if has_request_context() or _worker_chart.get() or not metadata_refresh_enabled(): + yield + return + # A task may have spent its fallback budget on earlier charts or other work. + # Restore that state after this chart, including on cancellation or failure. + operation_token: Token[MetadataOperation | None] = _worker_operation.set(None) + chart_token: Token[bool] = _worker_chart.set(True) + try: + with metadata_operation(): + yield + finally: + _worker_chart.reset(chart_token) + _worker_operation.reset(operation_token) + + +def metadata_refresh_enabled() -> bool: + return ( + has_app_context() + and current_app.config.get("SEMANTIC_LAYER_METADATA_REFRESH_ENABLED") is True + and is_feature_enabled("SEMANTIC_LAYERS") + ) + + +def _configuration(raw: str) -> dict[str, Any]: + """Cache parsing by stored bytes; providers receive independent mutable copies.""" + state: MetadataOperation | None = _current_operation() + if state is not None and raw in state.configurations: + return deepcopy(state.configurations[raw]) + parsed: Any = json.loads(raw) + if not isinstance(parsed, dict): + raise MetadataRefreshError("configuration") + if state is not None: + state.configurations[raw] = parsed + return deepcopy(parsed) + + +def participates(layer: SemanticLayer) -> bool: + """Classify stored configuration without leaking parser or registry errors.""" + if not metadata_refresh_enabled(): + return False + try: + configuration: dict[str, Any] = _configuration(layer.configuration) + return registry[layer.type].supports_metadata_refresh(configuration) + except (KeyError, TypeError, ValueError): + raise MetadataRefreshError("configuration") from None + + +def connection_metadata_scope(layer: SemanticLayer) -> str: + namespace: Any = current_app.config.get("SEMANTIC_LAYER_METADATA_NAMESPACE") + if callable(namespace): + namespace = namespace() + secret: Any = current_app.config.get("SECRET_KEY") + if isinstance(secret, bytes): + secret = secret.decode() + if ( + not isinstance(namespace, str) + or not isinstance(secret, str) + or layer.uuid is None + ): + raise MetadataRefreshError("configuration") + configuration: str = json.dumps( + {"provider": layer.type, "configuration": _configuration(layer.configuration)}, + sort_keys=True, + ) + return metadata_scope(secret, namespace, str(layer.uuid), configuration) + + +def connection_store(layer: SemanticLayer) -> ScopedMetadataStore: + """Resolve only stored, server-owned scope and recheck it before publication.""" + state: MetadataOperation = _operation(require_budget=False) + scope: str = connection_metadata_scope(layer) + if scope in state.stores: + return state.stores[scope] + operation_deadline() + config: Any = current_app.config.get("DISTRIBUTED_COORDINATION_CONFIG") + if not isinstance(config, dict): + raise MetadataRefreshError("unavailable") + try: + backend: DeadlineRedisBackend = DeadlineRedisBackend( + config, deadline=state.deadline + ) + except ValueError: + raise MetadataRefreshError("configuration") from None + + def revalidate() -> None: + # Avoid app-init regression: binding loads before encrypted model fields. + from superset.semantic_layers.models import SemanticLayer + + if not participates(layer): + raise MetadataRefreshError("configuration_changed") + session: Session + try: + with Session( + bind=db.session.get_bind(mapper=SemanticLayer), autoflush=False + ) as session: + fresh: SemanticLayer | None = session.get(SemanticLayer, layer.uuid) + if fresh is None or connection_metadata_scope(fresh) != scope: + raise MetadataRefreshError("configuration_changed") + except SQLAlchemyError: + logger.warning("Metadata layer revalidation failed", exc_info=True) + raise MetadataRefreshError("unavailable") from None + + store: ScopedMetadataStore = ScopedMetadataStore( + backend, + scope, + deadline=state.deadline, + snapshot_ttl_seconds=current_app.config[ + "SEMANTIC_LAYER_METADATA_SNAPSHOT_TTL_SECONDS" + ], + before_publish=revalidate, + ) + state.stores[scope] = store + return store + + +def layer_implementation(layer: SemanticLayer) -> LayerABC[Any, ViewABC]: + """Construct once per operation, binding the store before any discovery.""" + state: MetadataOperation = _operation(require_budget=False) + scope: str = connection_metadata_scope(layer) + if scope not in state.layers: + operation_deadline() + store: ScopedMetadataStore = connection_store(layer) + implementation: LayerABC[Any, ViewABC] = registry[ + layer.type + ].from_configuration(_configuration(layer.configuration)) Review Comment: If a provider's `from_configuration()` raises `ValueError`/`TypeError` (for example a stored configuration that no longer validates after a provider upgrade adds a required option), this path lets it escape. `RefreshMetadataCommand` turns the same failure into the typed `configuration` error (422), but chart-data, `/compatible` and datasource metadata requests that build the provider here return a generic 400/500 or a misclassified error, and the constructor's message can reach the response. Could construction failures be normalized to `MetadataRefreshError("configuration")` here so every discovery path maps them the same way? ########## tests/unit_tests/semantic_layers/metadata_result_inspection_test.py: ########## @@ -0,0 +1,315 @@ +# 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. + +from __future__ import annotations + +from typing import Any +from unittest.mock import Mock, patch + +import numpy as np +import pytest +from flask import Flask, g + +from superset.common.query_context import QueryContext +from superset.common.query_object import QueryObject +from superset.semantic_layers.cache_inspection import CacheEntryInfo +from superset.semantic_layers.models import SemanticView +from superset.semantic_layers.registry import registry +from tests.unit_tests.semantic_layers.metadata_identity_test import ( + context_for, + RefreshLayer, + ResultView, + view_for, +) + + [email protected]("total", [np.float32(12.5), np.int32(12)]) +def test_result_capture_excludes_runtime_contribution_totals( + app: Flask, monkeypatch: pytest.MonkeyPatch, total: Any +) -> None: + """Diagnostic capture shares the query key's runtime-total exclusion.""" + from superset.semantic_layers.result_inspection import captured_result_key + + provider: ResultView = ResultView("captured", 17) + context: QueryContext + query: QueryObject + context, query = context_for(view_for(provider)) + options: dict[str, Any] = {"columns": ["orders"], "contribution_totals": total} + rename_options: dict[str, Any] = {"columns": {"orders": "Order count"}} + query.post_processing = [ + {"operation": "contribution", "options": options}, + {"operation": "rename", "options": rename_options}, + ] + monkeypatch.setitem(app.config, "SEMANTIC_LAYER_METADATA_REFRESH_ENABLED", True) + manager: Mock = Mock() + manager.get_rls_cache_key.return_value = [] + with ( + app.test_request_context(), + patch( + "superset.semantic_layers.metadata_binding.participates", return_value=True + ), + patch( + "superset.semantic_layers.result_inspection.metadata_refresh_enabled", + return_value=True, + ), + patch( + "superset.semantic_layers.metadata_binding.view_implementation", + return_value=provider, + ), + patch("superset.semantic_layers.result_inspection.security_manager", manager), + patch("superset.common.query_context_processor.security_manager", manager), + ): + key: str | None = context.query_cache_key(query) + assert key is not None + assert captured_result_key(context, query) == key + assert options["contribution_totals"] is total + options["contribution_totals"] = np.float32(99.5) + assert captured_result_key(context, query) == key + assert rename_options == {"columns": {"orders": "Order count"}} + rename_options["columns"] = {"orders": "Renamed count"} + assert captured_result_key(context, query) is None + rename_options["columns"] = {"orders": "Order count"} + assert captured_result_key(context, query) == key + options["columns"] = ["revenue"] + assert captured_result_key(context, query) is None + + [email protected]("changed", ["subject", "query"]) +def test_result_capture_rejects_changed_subject_or_query( + app: Flask, changed: str +) -> None: + """A captured key never survives a subject or query edit with unchanged RLS.""" + from superset.semantic_layers.result_inspection import ( + capture_result_identity, + captured_result_key, + ) + + context: QueryContext + query: QueryObject + context, query = context_for(view_for(ResultView("captured", 17))) + manager: Mock = Mock() + manager.get_rls_cache_key.return_value = ["unchanged-rule"] + with ( + app.test_request_context(), + patch( + "superset.semantic_layers.result_inspection.metadata_refresh_enabled", + return_value=True, + ), + patch("superset.semantic_layers.result_inspection.security_manager", manager), + ): + g.user = Mock(id=1) + capture_result_identity(context, query, "existing-key") + assert captured_result_key(context, query) == "existing-key" + if changed == "subject": + g.user = Mock(id=2) + else: + query.metrics = ["revenue"] + assert captured_result_key(context, query) is None + + +def test_result_inspection_keeps_query_rls_and_never_constructs_provider( + app: Flask, monkeypatch: pytest.MonkeyPatch +) -> None: + from superset.commands.semantic_layer.inspect_query_result import ( + InspectQueryResultCommand, + ) + + provider: ResultView = ResultView("unused", 17) + view: SemanticView = view_for(provider) + monkeypatch.setitem(app.config, "SEMANTIC_LAYER_METADATA_REFRESH_ENABLED", True) + monkeypatch.setitem(registry, "cache-test", RefreshLayer) + context: QueryContext + query: QueryObject + context, query = context_for(view) + manager: Mock = Mock() + keys: list[str] = [] + inspect_entry: Mock + construct: Mock + rls: list[str] + with ( + app.test_request_context(), + patch( + "superset.semantic_layers.metadata_binding.is_feature_enabled", + return_value=True, + ), + patch( + "superset.commands.semantic_layer.inspect_query_result.security_manager", + manager, + ), + patch("superset.semantic_layers.result_inspection.security_manager", manager), + patch("superset.common.query_context_processor.security_manager", manager), + patch( + "superset.commands.semantic_layer.inspect_query_result.inspect_derived_entry" + ) as inspect_entry, + patch( + "superset.semantic_layers.metadata_binding.view_implementation", + return_value=provider, + ) as construct, + ): + assert InspectQueryResultCommand(context, 0).run().state == "unsupported" + construct.assert_not_called() + for rls in (["rule-a"], ["rule-b"]): + manager.get_rls_cache_key.return_value = rls + key: str | None = context.query_cache_key(query) + construct.reset_mock() + InspectQueryResultCommand(context, 0).run() + assert inspect_entry.call_args.args == (key, "query_result") + keys.append(inspect_entry.call_args.args[0]) + construct.assert_not_called() + assert keys[0] != keys[1] + manager.get_rls_cache_key.return_value = ["revoked"] + assert InspectQueryResultCommand(context, 0).run().state == "unsupported" + manager.raise_for_access.side_effect = PermissionError("denied") + inspect_entry.reset_mock() + with pytest.raises(PermissionError): + InspectQueryResultCommand(context, 0).run() + inspect_entry.assert_not_called() + + # A new request cannot reuse identities kept by an earlier request, even + # when an internal caller retains the same Python context object. + with ( + app.test_request_context(), + patch( + "superset.commands.semantic_layer.inspect_query_result.security_manager", + Mock(), + ), + ): + assert InspectQueryResultCommand(context, 0).run().state == "unsupported" Review Comment: This final block never reaches the cross-request check it describes. The `is_feature_enabled` patch from the earlier block has ended, so `metadata_refresh_enabled()` is false and `captured_result_key` returns `None` regardless of what is stored. If capture state moved onto the retained context, the first request's key would be reused by the second and this assertion would still pass. Could the second request run with the feature enabled and assert it reports `unsupported` and never reaches the derived-cache reader? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
