codeant-ai-for-open-source[bot] commented on code in PR #42760: URL: https://github.com/apache/superset/pull/42760#discussion_r3716181937
########## 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: **Suggestion:** The result value is written before descriptor registration, and a coordination failure returns without deleting that value. With a non-expiring timeout, every failed registration leaves an undiscoverable backend object indefinitely; even with expiration, repeated failures waste cache capacity. Delete `value_key` when mutation does not complete successfully. [resource leak] <details> <summary><b>Severity Level:</b> Major ⚠️</summary> ```mdx - ⚠️ Redis coordination failures create undiscoverable values. - ❌ Non-expiring semantic caches retain orphaned results indefinitely. - ⚠️ Repeated failed stores increase backend storage usage. ``` </details> [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=6c92efded7d14ff991691ad48995ecb5&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=6c92efded7d14ff991691ad48995ecb5&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) *(Use Cmd/Ctrl + Click for best experience)* <details> <summary><b>Prompt for AI Agent 🤖 </b></summary> ```mdx This is a comment left during a code review. **Path:** superset/semantic_layers/cache_repository.py **Line:** 175:176 **Comment:** *Resource Leak: The result value is written before descriptor registration, and a coordination failure returns without deleting that value. With a non-expiring timeout, every failed registration leaves an undiscoverable backend object indefinitely; even with expiration, repeated failures waste cache capacity. Delete `value_key` when mutation does not complete successfully. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42760&comment_hash=0cd60ec28d749203a789f94478b1c7f1c93c9dc455f33df43d392b6fcfc1e89d&reaction=like'>👍</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42760&comment_hash=0cd60ec28d749203a789f94478b1c7f1c93c9dc455f33df43d392b6fcfc1e89d&reaction=dislike'>👎</a> ########## 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: **Suggestion:** When the descriptor list exceeds the bound, older descriptors are discarded but their referenced result values are never deleted. Those values become permanently unreachable when `meta.timeout` is `None`, so storing many distinct queries can grow the backend without bound despite the descriptor limit. Remove the value objects corresponding to evicted descriptors or provide an explicit garbage-collection mechanism. [resource leak] <details> <summary><b>Severity Level:</b> Major ⚠️</summary> ```mdx - ⚠️ More than 128 queries evict descriptors. - ❌ Evicted result objects remain unreachable in cache storage. - ⚠️ Non-expiring buckets can grow with distinct queries. ``` </details> [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=5d29e382834546929f250d17db1b32b0&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=5d29e382834546929f250d17db1b32b0&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) *(Use Cmd/Ctrl + Click for best experience)* <details> <summary><b>Prompt for AI Agent 🤖 </b></summary> ```mdx This is a comment left during a code review. **Path:** superset/semantic_layers/cache_repository.py **Line:** 198:202 **Comment:** *Resource Leak: When the descriptor list exceeds the bound, older descriptors are discarded but their referenced result values are never deleted. Those values become permanently unreachable when `meta.timeout` is `None`, so storing many distinct queries can grow the backend without bound despite the descriptor limit. Remove the value objects corresponding to evicted descriptors or provide an explicit garbage-collection mechanism. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42760&comment_hash=dab94065c2516cebf9770844c1f81addc568c698e6fa3134e7f6356fe7160f0f&reaction=like'>👍</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42760&comment_hash=dab94065c2516cebf9770844c1f81addc568c698e6fa3134e7f6356fe7160f0f&reaction=dislike'>👎</a> -- 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]
