mikebridge commented on code in PR #42760: URL: https://github.com/apache/superset/pull/42760#discussion_r3834348313
########## superset/semantic_layers/cache_host.py: ########## @@ -0,0 +1,165 @@ +# 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 adaptation for semantic cache identity and provider contracts.""" + +import hashlib +from collections.abc import Iterable, Mapping +from datetime import datetime +from typing import cast, Protocol + +from flask import g, has_request_context +from superset_core.semantic_layers.layer import ( + SemanticCacheCapabilities, + SemanticCacheExecutionContext, + SemanticCacheIdentityMaterial, + SemanticCacheResponsibility, + SemanticCacheScope, +) + +from superset import security_manager +from superset.connectors.sqla.models import BaseDatasource +from superset.semantic_layers.cache_identity import ( + SemanticCacheIdentityFactory, + SemanticViewIdentity, +) +from superset.semantic_layers.cache_policy import ( + ContainmentCapabilities, + PatternSemantics, +) +from superset.semantic_layers.cache_repository import ViewMeta +from superset.utils import json + + +class _SemanticCacheProvider(Protocol): + semantic_cache_responsibility: SemanticCacheResponsibility + semantic_cache_scope: SemanticCacheScope + semantic_cache_capabilities: SemanticCacheCapabilities + + def get_semantic_cache_provider_identity( + self, + ) -> SemanticCacheIdentityMaterial | None: ... # pragma: no cover + + def get_semantic_cache_context_identity( + self, + context: SemanticCacheExecutionContext, + ) -> SemanticCacheIdentityMaterial | None: ... # pragma: no cover + + +def _execution_context( + datasource: BaseDatasource, +) -> SemanticCacheExecutionContext | None: + if not has_request_context() or not getattr(g, "user", None): + return None + user: object = g.user + principal: object = getattr(user, "id", None) or getattr(user, "username", None) + if principal is None: + return None + roles_value: object = getattr(user, "roles", ()) + roles: Iterable[object] = ( + roles_value + if isinstance(roles_value, Iterable) and not isinstance(roles_value, str) + else () + ) + role_ids: tuple[str, ...] = tuple( + sorted( + str(getattr(role, "id", None) or getattr(role, "name", "")) + for role in roles + ) + ) + guest_token: object = getattr(user, "guest_token", None) + rls_cache_key: list[str] = security_manager.get_rls_cache_key(datasource) + identity_payload: str = json.dumps( + { + "guest_token": guest_token, + "principal": str(principal), + "rls": rls_cache_key, + "roles": role_ids, + }, + default=str, + separators=(",", ":"), + sort_keys=True, + ) + host_identity: str = hashlib.sha256(identity_payload.encode()).hexdigest() + return SemanticCacheExecutionContext(str(principal), role_ids, host_identity) + + +def build_cache_configuration( + datasource: BaseDatasource, +) -> tuple[ViewMeta, ContainmentCapabilities] | None: + """Build safe host metadata when the provider explicitly opts in.""" + layer: _SemanticCacheProvider = cast( + _SemanticCacheProvider, + datasource.semantic_layer.implementation, + ) + if ( + getattr(layer, "semantic_cache_responsibility", None) + is not SemanticCacheResponsibility.SUPERSET + ): + return None + provider_material: object = layer.get_semantic_cache_provider_identity() + if not isinstance(provider_material, SemanticCacheIdentityMaterial): + return None + scope: object = getattr(layer, "semantic_cache_scope", None) + scope_material: Mapping[str, object] + if scope is SemanticCacheScope.GLOBAL: + scope_material = {"scope": "global"} + elif scope is SemanticCacheScope.EXECUTION_CONTEXT: + context: SemanticCacheExecutionContext | None = _execution_context(datasource) + if context is None: + return None + context_material: object = layer.get_semantic_cache_context_identity(context) + if not isinstance(context_material, SemanticCacheIdentityMaterial): + return None + scope_material = { + "host_identity": context.host_identity, + "provider_identity": context_material.values, + } + else: + return None + provider_capabilities: object = getattr(layer, "semantic_cache_capabilities", None) + if not isinstance(provider_capabilities, SemanticCacheCapabilities): + return None + pattern_semantics: PatternSemantics | None = ( + PatternSemantics.sql_like(escape=provider_capabilities.pattern_escape) + if provider_capabilities.pattern_escape is not None + else None + ) + capabilities: ContainmentCapabilities = ContainmentCapabilities( + comparisons=provider_capabilities.comparisons, + membership=provider_capabilities.membership, + nulls=provider_capabilities.nulls, + pattern_semantics=pattern_semantics, + ) + changed_on: object = getattr(datasource, "changed_on", None) + definition_material: dict[str, object] = { + "changed_on": changed_on.isoformat() + if isinstance(changed_on, datetime) + else str(changed_on), + } + meta: ViewMeta = ViewMeta( + view_identity=SemanticViewIdentity(str(datasource.uuid)), + definition_identity=SemanticCacheIdentityFactory.definition( + definition_material + ), + provider_identity=SemanticCacheIdentityFactory.provider( + provider_material.values + ), + scope_identity=SemanticCacheIdentityFactory.scope(scope_material), + timeout=datasource.cache_timeout, Review Comment: Confirmed and fixed in `e5d5077568`. `build_cache_configuration` now bypasses containment entirely when `cache_timeout == CACHE_DISABLED_TIMEOUT`, mirroring the guard in `set_and_log_cache` — the raw `-1` previously reached the backend, where Redis reads a negative TTL as *no expiry*, inverting "never cache" into "cache forever". Regression test `test_cache_configuration_bypasses_disabled_caching` fails against the previous code. *Fix authored with AI assistance; verified as described.* ########## superset/semantic_layers/cache_identity.py: ########## @@ -0,0 +1,316 @@ +# 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. + +"""Immutable identity values for semantic containment caching.""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping as MappingABC +from dataclasses import dataclass +from datetime import date, datetime, time, timedelta +from typing import Mapping + +from superset_core.semantic_layers.types import ( + AdhocExpression, + Dimension, + Filter, + GroupLimit, + Metric, + OrderTuple, + SemanticQuery, +) + +from superset.utils import json + +IDENTITY_FORMAT_VERSION: str = "v2" +_SENSITIVE_KEY_PARTS: tuple[str, ...] = ( + "credential", + "password", + "secret", + "token", +) + + +class SensitiveIdentityMaterialError(ValueError): + """Raised when raw secret-like material is offered for cache identity.""" + + +def _find_sensitive_paths(value: object, prefix: str = "") -> list[str]: + sensitive_paths: list[str] = [] + if isinstance(value, MappingABC): + for key, nested_value in value.items(): + key_text: str = str(key) + path: str = f"{prefix}.{key_text}" if prefix else key_text + if any(part in key_text.casefold() for part in _SENSITIVE_KEY_PARTS): + sensitive_paths.append(path) + sensitive_paths.extend(_find_sensitive_paths(nested_value, path)) + elif isinstance(value, (list, tuple)): + for index, nested_value in enumerate(value): + path = f"{prefix}[{index}]" + sensitive_paths.extend(_find_sensitive_paths(nested_value, path)) + return sensitive_paths + + +@dataclass(frozen=True) +class SemanticViewIdentity: + """Stable identity of a semantic view.""" + + value: str + + +@dataclass(frozen=True) +class SemanticDefinitionIdentity: + """Versioned digest of result-affecting semantic-view definition fields.""" + + digest: str + + +@dataclass(frozen=True) +class SemanticCacheProviderIdentity: + """Versioned digest of result-affecting provider configuration.""" + + digest: str + + +@dataclass(frozen=True) +class SemanticCacheScopeIdentity: + """Digest separating global or execution-context cache scopes.""" + + digest: str + + +class SemanticCacheIdentityFactory: + """Create deterministic, versioned identities from secret-safe material.""" + + @classmethod + def definition( + cls, + material: Mapping[str, object], + ) -> SemanticDefinitionIdentity: + """Create a semantic-definition identity.""" + return SemanticDefinitionIdentity(cls._digest(material)) + + @classmethod + def provider( + cls, + material: Mapping[str, object], + ) -> SemanticCacheProviderIdentity: + """Create a provider-configuration identity.""" + return SemanticCacheProviderIdentity(cls._digest(material)) + + @classmethod + def scope( + cls, + material: Mapping[str, object], + ) -> SemanticCacheScopeIdentity: + """Create an execution-scope identity.""" + return SemanticCacheScopeIdentity(cls._digest(material)) + + @classmethod + def query(cls, query: SemanticQuery) -> str: + """Create a logical identity from every result-affecting query field.""" + digest: str = cls._digest(_canonical_query(query)) + return f"semantic-cache:query:{digest}" + + @classmethod + def bucket( + cls, + view: SemanticViewIdentity, + definition: SemanticDefinitionIdentity, + provider: SemanticCacheProviderIdentity, + scope: SemanticCacheScopeIdentity, + ) -> str: + """Create a descriptor-bucket identity from all host boundaries.""" + digest: str = cls._digest( + { + "view": view.value, + "definition": definition.digest, + "provider": provider.digest, + "scope": scope.digest, + } + ) + return f"semantic-cache:bucket:{digest}" + + @classmethod + def value(cls, bucket: str, query: SemanticQuery) -> str: + """Create a result-value identity within a host boundary bucket.""" + digest: str = cls._digest({"bucket": bucket, "query": cls.query(query)}) + return f"semantic-cache:value:{digest}" + + @staticmethod + def order(order: list[OrderTuple] | None) -> str: + """Serialize ordering for descriptor compatibility checks.""" + if not order: + return "" + return _canonical_json( + [ + { + "element": _canonical_orderable(element), + "direction": direction.value, + } + for element, direction in order + ] + ) + + @staticmethod + def group_limit(group_limit: GroupLimit | None) -> str: + """Serialize group-limit behavior for descriptor compatibility checks.""" + if group_limit is None: + return "" + return _canonical_json(_canonical_group_limit(group_limit)) + + @staticmethod + def _digest(material: Mapping[str, object]) -> str: + sensitive_paths: list[str] = _find_sensitive_paths(material) + if sensitive_paths: + joined_keys: str = ", ".join(sorted(sensitive_paths)) + raise SensitiveIdentityMaterialError( + f"Secret-like identity fields are not allowed: {joined_keys}" + ) + + canonical: str = json.dumps( + material, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + payload: bytes = f"{IDENTITY_FORMAT_VERSION}:{canonical}".encode() + return f"{IDENTITY_FORMAT_VERSION}:{hashlib.sha256(payload).hexdigest()}" + + +def _canonical_query(query: SemanticQuery) -> Mapping[str, object]: Review Comment: Confirmed and fixed in `e5d5077568`, via bypass rather than key-widening: row-count dispatches now skip the cache in **both** directions (`cacheable=not query_object.is_rowcount` on the main and offset paths). Reasoning: containment reuse is only defined over tabular results — even with the result kind folded into the identity, a row-count entry could never survive `transform_result`'s column projection, so kind-scoped entries would only ever be dead weight in the descriptor bucket. Bypassing keeps the identity model simple and fixes both failure directions (row-count reusing table rows, and a stored scalar poisoning later table lookups). Regression test `test_rowcount_dispatch_bypasses_cache_both_directions` also pins that the tabular entry seeded by a prior table query stays undisturbed. *Fix authored with AI assistance; verified as described.* ########## superset/semantic_layers/cache.py: ########## @@ -0,0 +1,310 @@ +# 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. + +"""Application-level state for semantic containment caching.""" + +from __future__ import annotations + +import enum +import logging +from collections.abc import Callable +from dataclasses import dataclass +from typing import Protocol + +from superset_core.semantic_layers.types import SemanticQuery, SemanticResult + +from superset.semantic_layers.cache_coordination import ( + OwnerTokenCoordinationBackend, + SemanticCacheCoordinationSettings, + SemanticCacheCoordinator, +) +from superset.semantic_layers.cache_policy import ( + ContainmentCapabilities, + eligible_reuse_mode as eligible_reuse_mode, + rank_eligible_entries as rank_eligible_entries, +) +from superset.semantic_layers.cache_repository import ( + SemanticCacheBackend, + SemanticCacheBackendError, + SemanticCacheLookupError, + SemanticCacheLookupResult, + SemanticCacheRepository, + SemanticCacheStoreError, + ViewMeta, +) +from superset.semantic_layers.cache_transform import ( + SemanticCacheTransformationError, + transform_result, +) + + +class SemanticCacheDisabledReason(str, enum.Enum): + """Stable operational reasons why containment is ineffective.""" + + PARENT_FLAG_OFF = "parent_flag_off" + FLAG_OFF = "flag_off" + UNSUPPORTED_COORDINATION = "unsupported_coordination" + INVALID_COORDINATION_CONFIGURATION = "invalid_coordination_configuration" + + +@dataclass(frozen=True) +class SemanticCacheState: + """Requested and effective containment state owned by the cache service.""" + + requested: bool + effective: bool + disabled_reason: SemanticCacheDisabledReason | None + + def __post_init__(self) -> None: + if self.effective and not self.requested: + raise ValueError("Effective semantic caching must be requested") + if self.effective and self.disabled_reason is not None: + raise ValueError("Effective semantic caching cannot have a disabled reason") + if not self.effective and self.disabled_reason is None: + raise ValueError("Ineffective semantic caching requires a disabled reason") + + @classmethod + def enabled(cls) -> SemanticCacheState: + """Return the single valid effective state.""" + return cls(requested=True, effective=True, disabled_reason=None) + + @classmethod + def disabled( + cls, + reason: SemanticCacheDisabledReason, + *, + requested: bool = False, + ) -> SemanticCacheState: + """Return an explicitly ineffective state.""" + return cls( + requested=requested, + effective=False, + disabled_reason=reason, + ) + + +@dataclass(frozen=True) +class SemanticCacheOutcome: + """Host-side result plus whether containment supplied it.""" + + result: SemanticResult + cache_hit: bool + + +class SemanticCacheMetrics(Protocol): + """Fixed-name metrics used by containment cache operations.""" + + def incr(self, key: str) -> None: ... # pragma: no cover + + def gauge(self, key: str, value: float) -> None: ... # pragma: no cover + + +SEMANTIC_CACHE_METRIC_PREFIX: str = "semantic_cache.containment" +logger: logging.Logger = logging.getLogger(__name__) + + +class SafeSemanticCacheBackend: + """Translate pluggable cache-client failures into the repository boundary.""" + + def __init__(self, backend: SemanticCacheBackend) -> None: + self._backend: SemanticCacheBackend = backend + + def get(self, key: str) -> object | None: + try: + return self._backend.get(key) + except Exception as ex: # pylint: disable=broad-exception-caught + raise SemanticCacheBackendError("Semantic cache get failed") from ex + + def set( + self, + key: str, + value: object, + timeout: int | None = None, + ) -> bool: + try: + return self._backend.set(key, value, timeout=timeout) + except Exception as ex: # pylint: disable=broad-exception-caught + raise SemanticCacheBackendError("Semantic cache set failed") from ex + + def delete(self, key: str) -> bool: + try: + return self._backend.delete(key) + except Exception as ex: # pylint: disable=broad-exception-caught + raise SemanticCacheBackendError("Semantic cache delete failed") from ex + + +class SemanticCacheService: + """Orchestrate optional cache lookup, provider execution, and storage. + + This service is intentionally limited to application flow. Cache identity, + containment proofs, result transformation, persistence, distributed mutation, + and Superset host adaptation live in ``cache_identity``, ``cache_policy``, + ``cache_transform``, ``cache_repository``, ``cache_coordination``, and + ``cache_host`` respectively. Keeping those decisions behind narrow boundaries + prevents the orchestration service from becoming a multipurpose cache class. + """ + + def __init__( + self, + state: SemanticCacheState, + repository: SemanticCacheRepository | None = None, + metrics: SemanticCacheMetrics | None = None, + ) -> None: + if state.effective and repository is None: + raise ValueError("Effective semantic caching requires a repository") + self.state: SemanticCacheState = state + self._repository: SemanticCacheRepository | None = repository + self._metrics: SemanticCacheMetrics | None = metrics + + def _increment(self, suffix: str) -> None: + metrics: SemanticCacheMetrics | None = self._metrics + if metrics is None: + return + try: + metrics.incr(f"{SEMANTIC_CACHE_METRIC_PREFIX}.{suffix}") + except Exception: # pylint: disable=broad-exception-caught + logger.debug("Semantic cache metric emission failed", exc_info=True) + + @classmethod + def default_ineffective(cls) -> SemanticCacheService: + """Build the safe process default used before capability initialization.""" + return cls(SemanticCacheState.disabled(SemanticCacheDisabledReason.FLAG_OFF)) + + def execute( + self, + meta: ViewMeta, + query: SemanticQuery, + provider: Callable[[SemanticQuery], SemanticResult], + *, + capabilities: ContainmentCapabilities, + force: bool = False, + ) -> SemanticCacheOutcome: + """Serve a proven hit or execute and best-effort store provider output.""" + repository: SemanticCacheRepository | None = self._repository + if not self.state.effective or repository is None: + self._increment("bypass") + return SemanticCacheOutcome(provider(query), cache_hit=False) + if not force: + try: + lookup_result: SemanticCacheLookupResult = repository.lookup( + meta, query, capabilities + ) + except SemanticCacheLookupError: + self._increment("lookup_failure") + return SemanticCacheOutcome(provider(query), cache_hit=False) + for candidate in lookup_result.candidates: + try: + transformed: SemanticResult = transform_result( + candidate.result, + query, + candidate.decision, + capabilities, + ) + except SemanticCacheTransformationError: + self._increment("transform_failure") + continue + self._increment("hit") + return SemanticCacheOutcome(transformed, cache_hit=True) + try: + repository.prune_missing( + meta, + lookup_result.missing_value_keys, + ) + except SemanticCacheLookupError: + self._increment("prune_failure") + self._increment("miss") + else: + self._increment("bypass") + result: SemanticResult = provider(query) + try: + repository.store(meta, query, result) Review Comment: Confirmed and fixed in `e5d5077568` — and your instinct here was better than reported: the crash escapes the transformation-error boundary entirely (`AttributeError` was not in the caught tuple), so the second request against an empty result 500'd rather than missing. Two changes: normalization now happens **before** the store (the dispatcher is wrapped, so both the cached copy and the direct path see a coerced empty Arrow table), and `transform_result` treats `AttributeError` as an incompatible-entry miss so any entry stored by a pre-fix build degrades gracefully. Two regression tests: `test_none_results_are_normalized_before_store` and `test_poisoned_none_entry_degrades_to_miss` — the latter's control run reproduces the exact `AttributeError` at `cache_transform.py:203` against the previous code. *Fix authored with AI assistance; verified as described.* -- 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]
