mikebridge commented on code in PR #42760: URL: https://github.com/apache/superset/pull/42760#discussion_r3716293506
########## superset/semantic_layers/cache_repository.py: ########## @@ -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. + +"""Storage records for semantic containment caching.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from time import time +from typing import cast, Protocol + +from superset_core.semantic_layers.types import SemanticQuery, SemanticResult + +from superset.semantic_layers.cache_identity import ( + semantic_dimension_key, + SemanticCacheIdentityFactory, + SemanticCacheProviderIdentity, + SemanticCacheScopeIdentity, + SemanticDefinitionIdentity, + SemanticViewIdentity, +) +from superset.semantic_layers.cache_policy import rank_reuse_decisions +from superset.semantic_layers.cache_types import ( + CachedEntry as CachedEntry, + CachedResultCandidate as CachedResultCandidate, + ContainmentCapabilities, + ReuseDecision, + SemanticCacheLookupResult as SemanticCacheLookupResult, +) + +MAX_SEMANTIC_CACHE_DESCRIPTORS_PER_BUCKET: int = 128 + + +class SemanticCacheRepositoryError(RuntimeError): + """Base error for expected cache-adapter failures.""" + + +class SemanticCacheLookupError(SemanticCacheRepositoryError): + """Raised when an expected backend lookup operation fails.""" + + +class SemanticCacheStoreError(SemanticCacheRepositoryError): + """Raised when an expected backend store operation fails.""" + + +class SemanticCacheBackendError(RuntimeError): + """Expected operational failure raised by an expiring backend adapter.""" + + +class SemanticCacheCoordinationError(RuntimeError): + """Expected operational failure raised by a mutation coordinator.""" + + +@dataclass(frozen=True) +class ViewMeta: + """Identity and expiry inputs for one semantic-view cache bucket.""" + + view_identity: SemanticViewIdentity + definition_identity: SemanticDefinitionIdentity + provider_identity: SemanticCacheProviderIdentity + scope_identity: SemanticCacheScopeIdentity + timeout: int | None + + +class SemanticCacheBackend(Protocol): + """Expiring value/descriptor operations required by the repository.""" + + def get(self, key: str) -> object | None: ... # pragma: no cover + + def set( + self, key: str, value: object, timeout: int | None = None + ) -> bool: ... # pragma: no cover + + def delete(self, key: str) -> bool: ... # pragma: no cover + + +class SemanticCacheMutationCoordinator(Protocol): + """Boundary for ownership-safe descriptor mutation.""" + + def mutate( + self, key: str, operation: Callable[[], None] + ) -> bool: ... # pragma: no cover + + +class SemanticCacheRepository: + """Store expiring results and bounded descriptors behind injected ports.""" + + def __init__( + self, + backend: SemanticCacheBackend, + coordinator: SemanticCacheMutationCoordinator, + *, + clock: Callable[[], float] = time, + ) -> None: + self._backend: SemanticCacheBackend = backend + self._coordinator: SemanticCacheMutationCoordinator = coordinator + self._clock: Callable[[], float] = clock + + def _get( + self, + key: str, + error_type: type[SemanticCacheRepositoryError], + ) -> object | None: + try: + return self._backend.get(key) + except SemanticCacheBackendError as ex: + raise error_type("Semantic cache backend get failed") from ex + + def _set( + self, + key: str, + value: object, + timeout: int | None, + error_type: type[SemanticCacheRepositoryError], + ) -> None: + try: + persisted: bool = self._backend.set(key, value, timeout=timeout) + except SemanticCacheBackendError as ex: + raise error_type("Semantic cache backend set failed") from ex + if not persisted: + raise error_type("Semantic cache backend rejected set") + + def _mutate( + self, + key: str, + operation: Callable[[], None], + error_type: type[SemanticCacheRepositoryError], + ) -> bool: + try: + return self._coordinator.mutate(key, operation) + except SemanticCacheCoordinationError as ex: + raise error_type("Semantic cache descriptor mutation failed") from ex + + @staticmethod + def _bucket_key(meta: ViewMeta) -> str: + return SemanticCacheIdentityFactory.bucket( + meta.view_identity, + meta.definition_identity, + meta.provider_identity, + meta.scope_identity, + ) + + @staticmethod + def _entries(value: object | None) -> list[CachedEntry]: + if not isinstance(value, list) or not all( + isinstance(entry, CachedEntry) for entry in value + ): + return [] + return cast(list[CachedEntry], value) + + def store( + self, + meta: ViewMeta, + query: SemanticQuery, + result: SemanticResult, + ) -> bool: + """Store a TTL-bounded value and register its bounded descriptor.""" + bucket_key: str = self._bucket_key(meta) + value_key: str = SemanticCacheIdentityFactory.value(bucket_key, query) + self._set(value_key, result, meta.timeout, SemanticCacheStoreError) + descriptor: CachedEntry = CachedEntry( Review Comment: Fixed in f143574cc4. If descriptor coordination does not register the value, the repository deletes the undiscoverable result key. The regression test now asserts that a failed coordination attempt leaves no cached value behind. ########## superset/semantic_layers/cache_repository.py: ########## @@ -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. + +"""Storage records for semantic containment caching.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from time import time +from typing import cast, Protocol + +from superset_core.semantic_layers.types import SemanticQuery, SemanticResult + +from superset.semantic_layers.cache_identity import ( + semantic_dimension_key, + SemanticCacheIdentityFactory, + SemanticCacheProviderIdentity, + SemanticCacheScopeIdentity, + SemanticDefinitionIdentity, + SemanticViewIdentity, +) +from superset.semantic_layers.cache_policy import rank_reuse_decisions +from superset.semantic_layers.cache_types import ( + CachedEntry as CachedEntry, + CachedResultCandidate as CachedResultCandidate, + ContainmentCapabilities, + ReuseDecision, + SemanticCacheLookupResult as SemanticCacheLookupResult, +) + +MAX_SEMANTIC_CACHE_DESCRIPTORS_PER_BUCKET: int = 128 + + +class SemanticCacheRepositoryError(RuntimeError): + """Base error for expected cache-adapter failures.""" + + +class SemanticCacheLookupError(SemanticCacheRepositoryError): + """Raised when an expected backend lookup operation fails.""" + + +class SemanticCacheStoreError(SemanticCacheRepositoryError): + """Raised when an expected backend store operation fails.""" + + +class SemanticCacheBackendError(RuntimeError): + """Expected operational failure raised by an expiring backend adapter.""" + + +class SemanticCacheCoordinationError(RuntimeError): + """Expected operational failure raised by a mutation coordinator.""" + + +@dataclass(frozen=True) +class ViewMeta: + """Identity and expiry inputs for one semantic-view cache bucket.""" + + view_identity: SemanticViewIdentity + definition_identity: SemanticDefinitionIdentity + provider_identity: SemanticCacheProviderIdentity + scope_identity: SemanticCacheScopeIdentity + timeout: int | None + + +class SemanticCacheBackend(Protocol): + """Expiring value/descriptor operations required by the repository.""" + + def get(self, key: str) -> object | None: ... # pragma: no cover + + def set( + self, key: str, value: object, timeout: int | None = None + ) -> bool: ... # pragma: no cover + + def delete(self, key: str) -> bool: ... # pragma: no cover + + +class SemanticCacheMutationCoordinator(Protocol): + """Boundary for ownership-safe descriptor mutation.""" + + def mutate( + self, key: str, operation: Callable[[], None] + ) -> bool: ... # pragma: no cover + + +class SemanticCacheRepository: + """Store expiring results and bounded descriptors behind injected ports.""" + + def __init__( + self, + backend: SemanticCacheBackend, + coordinator: SemanticCacheMutationCoordinator, + *, + clock: Callable[[], float] = time, + ) -> None: + self._backend: SemanticCacheBackend = backend + self._coordinator: SemanticCacheMutationCoordinator = coordinator + self._clock: Callable[[], float] = clock + + def _get( + self, + key: str, + error_type: type[SemanticCacheRepositoryError], + ) -> object | None: + try: + return self._backend.get(key) + except SemanticCacheBackendError as ex: + raise error_type("Semantic cache backend get failed") from ex + + def _set( + self, + key: str, + value: object, + timeout: int | None, + error_type: type[SemanticCacheRepositoryError], + ) -> None: + try: + persisted: bool = self._backend.set(key, value, timeout=timeout) + except SemanticCacheBackendError as ex: + raise error_type("Semantic cache backend set failed") from ex + if not persisted: + raise error_type("Semantic cache backend rejected set") + + def _mutate( + self, + key: str, + operation: Callable[[], None], + error_type: type[SemanticCacheRepositoryError], + ) -> bool: + try: + return self._coordinator.mutate(key, operation) + except SemanticCacheCoordinationError as ex: + raise error_type("Semantic cache descriptor mutation failed") from ex + + @staticmethod + def _bucket_key(meta: ViewMeta) -> str: + return SemanticCacheIdentityFactory.bucket( + meta.view_identity, + meta.definition_identity, + meta.provider_identity, + meta.scope_identity, + ) + + @staticmethod + def _entries(value: object | None) -> list[CachedEntry]: + if not isinstance(value, list) or not all( + isinstance(entry, CachedEntry) for entry in value + ): + return [] + return cast(list[CachedEntry], value) + + def store( + self, + meta: ViewMeta, + query: SemanticQuery, + result: SemanticResult, + ) -> bool: + """Store a TTL-bounded value and register its bounded descriptor.""" + bucket_key: str = self._bucket_key(meta) + value_key: str = SemanticCacheIdentityFactory.value(bucket_key, query) + self._set(value_key, result, meta.timeout, SemanticCacheStoreError) + descriptor: CachedEntry = CachedEntry( + filters=frozenset(query.filters or set()), + dimension_keys=frozenset( + semantic_dimension_key(dimension) for dimension in query.dimensions + ), + metric_ids=frozenset(metric.id for metric in query.metrics), + limit=query.limit, + offset=query.offset or 0, + order_key=SemanticCacheIdentityFactory.order(query.order), + group_limit_key=SemanticCacheIdentityFactory.group_limit(query.group_limit), + value_key=value_key, + timestamp=self._clock(), + ) + + def register() -> None: + entries: list[CachedEntry] = self._entries( + self._get(bucket_key, SemanticCacheStoreError) + ) + retained: list[CachedEntry] = [ + entry for entry in entries if entry.value_key != value_key + ] + retained.append(descriptor) + bounded: list[CachedEntry] = sorted( + retained, + key=lambda entry: entry.timestamp, + reverse=True, + )[:MAX_SEMANTIC_CACHE_DESCRIPTORS_PER_BUCKET] Review Comment: Fixed in f143574cc4. After the bounded descriptor list is persisted, the repository deletes result keys for descriptors evicted from that list. The regression test stores 129 distinct queries and asserts that only the 128 referenced result values remain. -- 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]
