sadpandajoe commented on code in PR #42760: URL: https://github.com/apache/superset/pull/42760#discussion_r3822170920
########## 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: The cache key only encodes `SemanticQuery`, while this caller dispatches the same query to either `get_table` or `get_row_count`. When a server-paginated table issues otherwise identical data and row-count requests, the row-count lookup can reuse table rows and return no `rowcount`, breaking pagination. Can the dispatched result kind be part of the cache identity? ########## 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: A datasource timeout of `-1` means caching is disabled, but this passes `-1` to the containment backend instead. RedisCache treats that as no expiry, so a query from a datasource configured never to cache can be reused indefinitely. Should this bypass containment (and use the resolved query timeout) when caching is disabled? ########## 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: This stores the provider result before `mapper.py` normalizes `SemanticResult(results=None)` to an empty Arrow table. The first empty result succeeds, but the next cache hit calls `result.results.select(...)` and raises on `None`. Could the result be normalized before it is stored, or skipped when it has no table? -- 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]
