mikebridge commented on code in PR #42760: URL: https://github.com/apache/superset/pull/42760#discussion_r4009553027
########## superset/semantic_layers/cache_coordination.py: ########## @@ -0,0 +1,207 @@ +# 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. + +"""Ownership-safe coordination for semantic-cache descriptor mutations.""" + +import logging +import math +from collections.abc import Callable +from dataclasses import dataclass +from random import random +from threading import Event, Thread +from time import monotonic, sleep +from typing import Protocol, runtime_checkable +from uuid import uuid4 + +from redis.exceptions import RedisError + +from superset.semantic_layers.cache_repository import SemanticCacheCoordinationError + +SEMANTIC_CACHE_COORDINATION_FAILURE_METRIC: str = ( + "semantic_cache.containment.coordination_failure" +) +logger: logging.Logger = logging.getLogger(__name__) + + +@runtime_checkable +class OwnerTokenCoordinationBackend(Protocol): + """Atomic lease operations required from a coordination backend.""" + + def acquire_owner_token( + self, + key: str, + owner_token: str, + lease_seconds: int, + ) -> bool: ... # pragma: no cover + + def release_owner_token( + self, + key: str, + owner_token: str, + ) -> bool: ... # pragma: no cover + + def refresh_owner_token( + self, + key: str, + owner_token: str, + lease_seconds: int, + ) -> bool: ... # pragma: no cover + + +@dataclass(frozen=True) +class SemanticCacheCoordinationSettings: + """Validated bounded timing settings for descriptor leases.""" + + wait_seconds: float + lease_seconds: int + + def __post_init__(self) -> None: + if ( + isinstance(self.wait_seconds, bool) + or not isinstance(self.wait_seconds, (int, float)) + or not math.isfinite(self.wait_seconds) + or self.wait_seconds < 0 + ): + raise ValueError( + "Coordination wait seconds must be finite and non-negative" + ) + if ( + isinstance(self.lease_seconds, bool) + or not isinstance(self.lease_seconds, int) + or self.lease_seconds <= 0 + ): + raise ValueError("Coordination lease seconds must be positive") + + +class SemanticCacheCoordinator: + """Run descriptor mutations while holding an owner-token lease.""" + + def __init__( + self, + backend: OwnerTokenCoordinationBackend, + settings: SemanticCacheCoordinationSettings, + *, + clock: Callable[[], float] = monotonic, + sleeper: Callable[[float], None] = sleep, + token_factory: Callable[[], str] = lambda: uuid4().hex, + failure_metric: Callable[[str], None] | None = None, + jitter: Callable[[], float] = random, + ) -> None: + self._backend: OwnerTokenCoordinationBackend = backend + self._settings: SemanticCacheCoordinationSettings = settings + self._clock: Callable[[], float] = clock + self._sleeper: Callable[[float], None] = sleeper + self._token_factory: Callable[[], str] = token_factory + self._failure_metric: Callable[[str], None] = failure_metric or (lambda _: None) + self._jitter: Callable[[], float] = jitter + + def _failure(self, message: str, cause: RedisError | None = None) -> None: + self._record_failure() + error: SemanticCacheCoordinationError = SemanticCacheCoordinationError(message) + if cause is None: + raise error + raise error from cause + + def _record_failure(self) -> None: + try: + self._failure_metric(SEMANTIC_CACHE_COORDINATION_FAILURE_METRIC) + except Exception: # pylint: disable=broad-exception-caught + logger.debug("Semantic cache coordination metric failed", exc_info=True) + + def _run_with_renewal( + self, + lease_key: str, + owner_token: str, + operation: Callable[[], None], + ) -> bool: + renewal_stopped: Event = Event() + renewal_failed: Event = Event() + + def renew_lease() -> None: + interval: float = self._settings.lease_seconds / 3 + while not renewal_stopped.wait(interval): + try: + refreshed: bool = self._backend.refresh_owner_token( + lease_key, + owner_token, + self._settings.lease_seconds, + ) + except RedisError: + renewal_failed.set() + return + if not refreshed: + renewal_failed.set() + return + + renewal_thread: Thread = Thread( + target=renew_lease, + name="semantic-cache-lease-renewal", + daemon=True, + ) + renewal_thread.start() + try: + operation() Review Comment: addressed in ab42028cf9 — the descriptor/value commit is now fenced by the owner token. A `compare_owner_and_set` Lua primitive on `RedisCommandsMixin` rejects the SET unless the lease key still holds the caller's token; it runs only when the value store and the lease share the exact same Redis client, and separate clients fall back to a documented, non-atomic immediate ownership recheck before the SET. A paused writer whose lease was taken over mid-operation now has its SET rejected, so the fresher payload and its descriptor/TTL survive (test: `test_same_client_fence_rejects_paused_writer`). *Posted by Claude (AI) on behalf of @mikebridge.* ########## superset/semantic_layers/cache_coordination.py: ########## @@ -0,0 +1,207 @@ +# 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. + +"""Ownership-safe coordination for semantic-cache descriptor mutations.""" + +import logging +import math +from collections.abc import Callable +from dataclasses import dataclass +from random import random +from threading import Event, Thread +from time import monotonic, sleep +from typing import Protocol, runtime_checkable +from uuid import uuid4 + +from redis.exceptions import RedisError + +from superset.semantic_layers.cache_repository import SemanticCacheCoordinationError + +SEMANTIC_CACHE_COORDINATION_FAILURE_METRIC: str = ( + "semantic_cache.containment.coordination_failure" +) +logger: logging.Logger = logging.getLogger(__name__) + + +@runtime_checkable +class OwnerTokenCoordinationBackend(Protocol): + """Atomic lease operations required from a coordination backend.""" + + def acquire_owner_token( + self, + key: str, + owner_token: str, + lease_seconds: int, + ) -> bool: ... # pragma: no cover + + def release_owner_token( + self, + key: str, + owner_token: str, + ) -> bool: ... # pragma: no cover + + def refresh_owner_token( + self, + key: str, + owner_token: str, + lease_seconds: int, + ) -> bool: ... # pragma: no cover + + +@dataclass(frozen=True) +class SemanticCacheCoordinationSettings: + """Validated bounded timing settings for descriptor leases.""" + + wait_seconds: float + lease_seconds: int + + def __post_init__(self) -> None: + if ( + isinstance(self.wait_seconds, bool) + or not isinstance(self.wait_seconds, (int, float)) + or not math.isfinite(self.wait_seconds) + or self.wait_seconds < 0 + ): + raise ValueError( + "Coordination wait seconds must be finite and non-negative" + ) + if ( + isinstance(self.lease_seconds, bool) + or not isinstance(self.lease_seconds, int) + or self.lease_seconds <= 0 + ): + raise ValueError("Coordination lease seconds must be positive") + + +class SemanticCacheCoordinator: + """Run descriptor mutations while holding an owner-token lease.""" + + def __init__( + self, + backend: OwnerTokenCoordinationBackend, + settings: SemanticCacheCoordinationSettings, + *, + clock: Callable[[], float] = monotonic, + sleeper: Callable[[float], None] = sleep, + token_factory: Callable[[], str] = lambda: uuid4().hex, + failure_metric: Callable[[str], None] | None = None, + jitter: Callable[[], float] = random, + ) -> None: + self._backend: OwnerTokenCoordinationBackend = backend + self._settings: SemanticCacheCoordinationSettings = settings + self._clock: Callable[[], float] = clock + self._sleeper: Callable[[float], None] = sleeper + self._token_factory: Callable[[], str] = token_factory + self._failure_metric: Callable[[str], None] = failure_metric or (lambda _: None) + self._jitter: Callable[[], float] = jitter + + def _failure(self, message: str, cause: RedisError | None = None) -> None: + self._record_failure() + error: SemanticCacheCoordinationError = SemanticCacheCoordinationError(message) + if cause is None: + raise error + raise error from cause + + def _record_failure(self) -> None: + try: + self._failure_metric(SEMANTIC_CACHE_COORDINATION_FAILURE_METRIC) + except Exception: # pylint: disable=broad-exception-caught + logger.debug("Semantic cache coordination metric failed", exc_info=True) + + def _run_with_renewal( + self, + lease_key: str, + owner_token: str, + operation: Callable[[], None], + ) -> bool: + renewal_stopped: Event = Event() + renewal_failed: Event = Event() + + def renew_lease() -> None: + interval: float = self._settings.lease_seconds / 3 + while not renewal_stopped.wait(interval): + try: + refreshed: bool = self._backend.refresh_owner_token( + lease_key, + owner_token, + self._settings.lease_seconds, + ) + except RedisError: + renewal_failed.set() + return + if not refreshed: + renewal_failed.set() + return + + renewal_thread: Thread = Thread( + target=renew_lease, + name="semantic-cache-lease-renewal", + daemon=True, + ) + renewal_thread.start() + try: + operation() + finally: + renewal_stopped.set() + renewal_thread.join() + return renewal_failed.is_set() + + def mutate(self, key: str, operation: Callable[[], None]) -> bool: + """Acquire a bounded lease, mutate, and release only our ownership.""" + lease_key: str = f"semantic-cache-lock:{key}" + owner_token: str = self._token_factory() + deadline: float = self._clock() + self._settings.wait_seconds + acquired: bool = False + while True: + try: + acquired = self._backend.acquire_owner_token( Review Comment: addressed in ab42028cf9 — coordination Redis I/O is now bounded. A private coordination client is constructed with socket/connect timeouts derived from `SEMANTIC_CACHE_COORDINATION_WAIT_SECONDS` (1 ms floor) and retry/backoff disabled, leaving the shared data-cache client untouched. A hung store/refresh/release is caught as a best-effort failure (coordination_failure + store_failure counters) and the chart still returns the provider result within the bound (test: `test_coordination_timeout_keeps_provider_result`). *Posted by Claude (AI) on behalf of @mikebridge.* ########## superset-frontend/src/explore/components/ChartPills.tsx: ########## @@ -100,12 +100,19 @@ export const ChartPills = forwardRef( limit={Number(rowLimit ?? 0)} /> )} - {!isLoading && firstQueryResponse?.is_cached && ( - <CachedLabel - onClick={refreshCachedQuery} - cachedTimestamp={firstQueryResponse.cached_dttm} - /> - )} + {!isLoading && + (firstQueryResponse?.is_cached || Review Comment: addressed in c25fcfaed6 — provenance now aggregates across every query response: all HIT → HIT, any HIT mixed with a non-HIT → MIXED (shown as "Mixed cache", not a semantic-cache hit), otherwise MISS. The server-paginated data-HIT + row-count-MISS case is pinned in `ChartPills.test.tsx`. *Posted by Claude (AI) on behalf of @mikebridge.* ########## docs/developer_docs/extensions/contribution-types.md: ########## @@ -291,3 +291,99 @@ class MySemanticLayer(SemanticLayer[MyConfig, MySemanticView]): - **Host context**: Original ID used as-is The decorator registers the class in the semantic layers registry, making it available in the UI for users to create connections. The `configuration_class` should be a Pydantic model that defines the fields needed to connect (credentials, project, database, etc.). Superset uses the model's JSON schema to render the configuration form dynamically. + +#### Semantic result containment caching + +Superset containment caching is experimental and off by default. A provider must +explicitly opt in; the safe defaults leave caching with the provider and scope +results to an execution context: + +```python +from superset_core.semantic_layers.layer import ( + SemanticCacheCapabilities, + SemanticCacheExecutionContext, + SemanticCacheIdentityMaterial, + SemanticCacheResponsibility, + SemanticCacheScope, +) + +class MySemanticLayer(SemanticLayer[MyConfig, MySemanticView]): + semantic_cache_responsibility = SemanticCacheResponsibility.SUPERSET + semantic_cache_scope = SemanticCacheScope.EXECUTION_CONTEXT + semantic_cache_capabilities = SemanticCacheCapabilities( + comparisons=True, + membership=True, + nulls=True, + pattern_escape="\\", + ) + + def get_semantic_cache_provider_identity(self) -> SemanticCacheIdentityMaterial: + return SemanticCacheIdentityMaterial( + {"provider_version": "v1", "catalog": self.config.catalog} + ) + + def get_semantic_cache_context_identity( + self, + context: SemanticCacheExecutionContext, + ) -> SemanticCacheIdentityMaterial: + return SemanticCacheIdentityMaterial({"tenant": self.tenant_id(context)}) +``` + +Identity material must be secret-free and include every provider setting that can +change results. For execution-context scope, Superset also hashes the principal, +roles, guest-token claims, and row-level-security cache key. Returning `None` from +either identity method bypasses containment. Use `GLOBAL` only when results are +provably identical across principals and tenants; containment is bypassed for a +`GLOBAL` view whenever Superset row-level security applies to the request, since +that variation is invisible to the provider. Declare only filter capabilities +whose provider semantics exactly match Superset's post-processing semantics. + +Operators enable both `SEMANTIC_LAYERS` and the development feature flag +`SEMANTIC_LAYER_CONTAINMENT_CACHE`, and configure two backends: + +- `DATA_CACHE_CONFIG` holds the cached results and their descriptors. It must be + a persistent cache shared by every web and worker process, such as `RedisCache`. + The default `NullCache` discards every value, so containment would only ever + miss; `RedisSentinelCache` reads from replicas that can lag the master. Both + disable containment at startup rather than run it ineffectively. +- `DISTRIBUTED_COORDINATION_CONFIG` must select `RedisCache` or + `RedisSentinelCache`; containment requires its atomic owner-token lease + operations. + +```python +DATA_CACHE_CONFIG = { + "CACHE_TYPE": "RedisCache", + "CACHE_DEFAULT_TIMEOUT": 86400, + "CACHE_KEY_PREFIX": "superset_results", + "CACHE_REDIS_URL": "redis://redis:6379/1", +} +DISTRIBUTED_COORDINATION_CONFIG = { + "CACHE_TYPE": "RedisCache", + "CACHE_REDIS_URL": "redis://redis:6379/2", +} +``` + +Containment follows the same cache timeout the ordinary result cache resolves for +a request (custom, then chart, then dataset), so a chart-level timeout of `-1` +bypasses containment entirely, and honors `DATA_CACHE_MAX_VALUE_SIZE`: a result +larger than that many bytes is served but not stored. The bounded defaults are +`SEMANTIC_CACHE_COORDINATION_WAIT_SECONDS = 1.0` and +`SEMANTIC_CACHE_COORDINATION_LEASE_SECONDS = 30`. A missing or unsuitable backend +and missing or invalid coordination each disable only containment, log a fixed +warning without configuration or query content, and leave provider execution +available. + +Roll out to a canary worker cohort after recording provider latency, error rate, +and cache-backend health for a comparable baseline. Observe at least one normal +traffic cycle. Alert ownership should sit with the semantic-platform operator. +Track `semantic_cache.containment.enabled` and the fixed-name `hit`, `miss`, Review Comment: addressed in c25fcfaed6 — the rollout guide now lists all twelve emitted counters with their full `semantic_cache.containment.` prefix (enabled, hit, miss, bypass, store_skipped, store_failure, lookup_failure, transform_failure, prune_failure, coordination_failure, unsupported, invalid_configuration) and calls out the failure counters to alert on. *Posted by Claude (AI) on behalf of @mikebridge.* -- 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]
