sadpandajoe commented on code in PR #44835: URL: https://github.com/apache/superset/pull/44835#discussion_r4172101395
########## superset/coordination/deadline_backend.py: ########## @@ -0,0 +1,343 @@ +# 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. + +"""Private, cancellation-bounded Redis operations for synchronous metadata callers. + +Each command owns its async client and loop. Cancelling the command disconnects +its socket; shared coordinator pools and their retry/timeout policy are untouched. +""" + +from __future__ import annotations + +import asyncio +import math +import os +import time +from _thread import LockType +from contextlib import AsyncExitStack +from typing import Any, TYPE_CHECKING + +from redis.asyncio import Redis +from redis.asyncio.retry import Retry +from redis.asyncio.sentinel import Sentinel +from redis.backoff import NoBackoff +from redis.exceptions import RedisError, TimeoutError as RedisTimeoutError +from superset_core.semantic_layers.metadata import ( + MetadataRefreshError, + remaining_budget, +) + +from superset.coordination.cache_backend import _COMPARE_AND_DELETE_LUA + +if TYPE_CHECKING: + from threading import local + + from gevent.threadpool import ThreadPool + +_METADATA_POOL_SIZE: int = 4 +_metadata_pool_state: local | None = None + + +def _metadata_threadpool() -> ThreadPool: + """Reuse a lazy metadata-only pool in its owning native thread and hub.""" + from gevent import get_hub + from gevent.hub import Hub + from gevent.monkey import get_original + from gevent.threadpool import ThreadPool + + global _metadata_pool_state # pylint: disable=global-statement + if _metadata_pool_state is None: + # Patched threading.local is greenlet-local; the pool belongs to a hub. + _metadata_pool_state = get_original("threading", "local")() + hub: Hub = get_hub() + pool: ThreadPool | None = getattr(_metadata_pool_state, "pool", None) + if pool is None or pool.hub is not hub or pool.pid != os.getpid(): + pool = ThreadPool(_METADATA_POOL_SIZE, hub=hub, idle_task_timeout=30) + _metadata_pool_state.pool = pool + return pool + + +_COMPARE_AND_PUBLISH_LUA: str = """ +if redis.call('get', KEYS[1]) ~= ARGV[1] then + return 0 +end +redis.call('psetex', KEYS[2], ARGV[3], ARGV[2]) +redis.call('del', KEYS[1]) +return 1 +""" + +_GET_WITH_TTL_LUA: str = """ +return {redis.call('get', KEYS[1]), redis.call('pttl', KEYS[1])} +""" + +_GET_OR_CREATE_LUA: str = """ +local value = redis.call('get', KEYS[1]) +if value then + return value +end +redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) +return ARGV[1] +""" + + +class _CommandCancellation: + """Transfer caller cancellation to its private native-thread asyncio task.""" + + def __init__(self, lock: LockType) -> None: + self._lock: LockType = lock + self._cancelled: bool = False + self._loop: asyncio.AbstractEventLoop | None = None + self._task: asyncio.Task[Any] | None = None + + def bind(self, loop: asyncio.AbstractEventLoop, task: asyncio.Task[Any]) -> None: + """A cancelled queued command must not begin Redis work.""" + with self._lock: + self._loop, self._task = loop, task + if self._cancelled: + task.cancel() + + def cancel(self) -> None: + """Schedule cancellation without calling task methods across threads.""" + with self._lock: + self._cancelled = True + if self._loop is not None and self._task is not None: + self._loop.call_soon_threadsafe(self._task.cancel) + + def detach(self) -> None: + """Prevent racing cancellation from targeting a closed loop.""" + with self._lock: + self._loop, self._task = None, None + + +class DeadlineRedisBackend: + """Use the coordinator configuration without sharing mutable connections.""" + + def __init__(self, config: dict[str, Any], *, deadline: float) -> None: + if not math.isfinite(deadline) or config.get("CACHE_TYPE") not in { + "RedisCache", + "RedisSentinelCache", + }: + raise ValueError("Unsupported metadata coordination configuration") + self._config: dict[str, Any] = dict(config) + self._deadline: float = deadline + + def with_deadline(self, deadline: float) -> DeadlineRedisBackend: + """Create a private call budget without extending the operation ceiling.""" + if not math.isfinite(deadline): + raise ValueError("Metadata deadline must be finite") + return DeadlineRedisBackend( + self._config, deadline=min(self._deadline, deadline) + ) + + def _remaining(self) -> float: + """Keep the transport's Redis error boundary while sharing SDK validation.""" + try: + return remaining_budget(self._deadline, now=time.monotonic()) + except MetadataRefreshError: + raise RedisTimeoutError("Metadata deadline invalid or expired") from None + + def _socket_timeout(self, key: str, remaining: float) -> float: + """Preserve shorter node timeouts inside the shared operation deadline.""" + configured: object = self._config.get(key) + if ( + isinstance(configured, (int, float)) + and not isinstance(configured, bool) + and 0 < configured < math.inf + ): + return min(configured, remaining) + return remaining + + async def _command(self, *args: str | int) -> Any: + remaining: float = self._remaining() + options: dict[str, Any] = { + "db": self._config.get("CACHE_REDIS_DB", 0), + "username": self._config.get("CACHE_REDIS_USER"), + "password": self._config.get("CACHE_REDIS_PASSWORD"), + "socket_timeout": self._socket_timeout( + "CACHE_REDIS_SOCKET_TIMEOUT", remaining + ), + "socket_connect_timeout": self._socket_timeout( + "CACHE_REDIS_SOCKET_CONNECT_TIMEOUT", remaining + ), + "retry": Retry(NoBackoff(), 0), + "protocol": 2, + } + if self._config.get("CACHE_REDIS_SSL", False): + options.update( + { + "ssl": True, + "ssl_certfile": self._config.get("CACHE_REDIS_SSL_CERTFILE"), + "ssl_keyfile": self._config.get("CACHE_REDIS_SSL_KEYFILE"), + "ssl_ca_certs": self._config.get("CACHE_REDIS_SSL_CA_CERTS"), + "ssl_cert_reqs": self._config.get( + "CACHE_REDIS_SSL_CERT_REQS", "required" + ), + } + ) + # One cancellation deadline covers DNS, Sentinel discovery, authentication, + # response parsing (including trickled responses) and connection cleanup. + stack: AsyncExitStack + async with asyncio.timeout(remaining), AsyncExitStack() as stack: + client: Redis + if self._config["CACHE_TYPE"] == "RedisSentinelCache": + sentinel: Sentinel = Sentinel( + self._config.get("CACHE_REDIS_SENTINELS", [("127.0.0.1", 26379)]), + sentinel_kwargs={ + "password": self._config.get("CACHE_REDIS_SENTINEL_PASSWORD"), + "socket_timeout": options["socket_timeout"], + "socket_connect_timeout": options["socket_connect_timeout"], + "retry": Retry(NoBackoff(), 0), + "protocol": 2, + }, + **options, + ) + sentinel_client: Redis + for sentinel_client in sentinel.sentinels: + stack.push_async_callback( + getattr(sentinel_client, "aclose", None) + or sentinel_client.close + ) + client = sentinel.master_for( + self._config.get("CACHE_REDIS_SENTINEL_MASTER", "mymaster") + ) + else: + client = Redis( + host=self._config.get("CACHE_REDIS_HOST", "localhost"), + port=self._config.get("CACHE_REDIS_PORT", 6379), + **options, + ) + await stack.enter_async_context(client) Review Comment: With supported redis-py 5.0.0, the Sentinel master client treats its supplied pool as externally owned, so this context-manager exit does not disconnect that private pool before its event loop closes; repeated commands can leave sockets awaiting garbage collection. Could the master pool be explicitly closed, with a minimum-version test that checks disconnection rather than mocking context-manager exit? ########## superset/tasks/async_queries.py: ########## @@ -178,30 +181,63 @@ def _inject_contribution_totals( cached dataframe and inject the sums into this query's contribution post-processing before it runs — the same result the synchronous ``ensure_totals_available`` produces, but reading the cache the prerequisite - populated instead of re-running the totals query. ``contribution_totals`` is + populated. Participating semantic queries resolve the totals context under + the dependent operation so an obsolete denominator is recomputed. + ``contribution_totals`` is stripped from the cache key, so this affects only the result, not the key. """ from superset.common.query_context_processor import is_summable from superset.common.utils.query_cache_manager import QueryCacheManager - cache = QueryCacheManager.get(key=totals_cache_key, region=CacheRegion.DATA) - if not cache.is_loaded or cache.df is None: - # The depends_on prerequisite guarantees the totals task succeeded and wrote - # this cache entry, so a miss is unexpected (e.g. it was evicted between the - # totals task finishing and this task reading). Fail loudly rather than - # caching a silently un-normalized result the client would then re-request: - # this task's single query cannot reproduce the synchronous path's - # ensure_totals_available (it has no totals query to run). - raise SupersetException( - f"Contribution totals not found in cache under {totals_cache_key}" + df: pd.DataFrame + if totals_context is not None: + # The ordinary key captures this operation's catalog. A matching totals + # entry is reused; a refreshed catalog recomputes before normalization. + totals_context.force = ( + totals_context.force + and _query_task_cache_key(totals_context, 0) != totals_cache_key ) - df = cache.df - totals = {col: df[col].sum() for col in df.columns if is_summable(df[col])} + totals_context.is_async_execution = True + with _capture_query_cancellation(totals_context): + df = totals_context.get_df_payload_result( + totals_context.queries[0] + ).payload["df"] Review Comment: A failed totals recomputation returns an empty DataFrame with status FAILED and an error, so this accepts it and injects empty contribution totals; normalization then uses only the returned main-query rows and can cache 100% where the full denominator would give 10%. Could the result status/error be checked before executing or caching the dependent query? ########## superset/semantic_layers/metadata_cache.py: ########## @@ -0,0 +1,109 @@ +# 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. + +"""Captured catalog identity for derived caches and read-only inspection.""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field +from typing import TYPE_CHECKING +from uuid import uuid4 + +from superset_core.semantic_layers.metadata import CatalogSnapshot, MetadataRefreshError + +from superset.semantic_layers import metadata_binding +from superset.semantic_layers.metadata import ScopedMetadataStore +from superset.semantic_layers.metadata_binding import connection_store +from superset.utils import json + +if TYPE_CHECKING: + from superset.semantic_layers.models import SemanticView + + +@dataclass(frozen=True) +class CompatibilityIdentity: + key: str = field(repr=False) + source_observed_at: str | None + + +def view_cache_token(view: SemanticView, token: str) -> str: + """Include host view configuration without revealing it in cache keys.""" + identity: str = json.dumps( + [token, str(view.uuid), view.name, json.loads(view.configuration)], + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + return hashlib.sha256(identity.encode()).hexdigest() + + +def annotation_cache_token(view: SemanticView) -> str | None: + """Key a host chart without discovering its annotation source's metadata.""" + if not metadata_binding.participates(view.semantic_layer): + return None + try: + token: str | None = metadata_binding.peek_view_metadata_token(view) Review Comment: On a cache miss, a SQL-backed parent that runs past the 30-second metadata budget cannot fetch its participating semantic annotation afterward: this lookup only peeked the catalog, so first view binding fails on the expired deadline even with warm metadata. Could the annotation observation be captured on a miss before the slow parent query starts? ########## superset/semantic_layers/metadata.py: ########## @@ -0,0 +1,442 @@ +# 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. + +"""Scoped publication and invalidation of provider-owned metadata.""" + +from __future__ import annotations + +import hashlib +import hmac +import math +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Literal, Protocol, TYPE_CHECKING +from uuid import uuid4 + +from redis.exceptions import RedisError +from superset_core.semantic_layers.metadata import ( + CatalogLoader, + CatalogSnapshot, + MetadataRefreshError, + MetadataRefreshResult, + remaining_budget, +) + +from superset.semantic_layers.cache_inspection import CacheEntryInfo, describe_entry +from superset.utils import json + +CATALOG_TTL_SECONDS: int = 300 +REFRESH_LEASE_SECONDS: int = 60 +FETCH_DEADLINE_SECONDS: int = 30 +MAX_CATALOG_BYTES: int = 10 * 1024 * 1024 +SNAPSHOT_FORMAT_VERSION: int = 2 +READER_POLL_SECONDS: float = 0.05 + + +if TYPE_CHECKING: + + class PublicationBackend(Protocol): + """The shared coordinator operations used by semantic metadata.""" + + def with_deadline(self, deadline: float) -> PublicationBackend: ... + def get(self, name: str) -> bytes | None: ... + def set( + self, + name: str, + value: str, + ex: int | None = None, + px: int | None = None, + nx: bool = False, + xx: bool = False, + ) -> bool | None: ... + def delete(self, *names: str) -> int: ... + def compare_and_delete(self, name: str, expected: str) -> int: ... + def compare_and_publish( + self, + lease_key: str, + expected: str, + snapshot_key: str, + value: str, + ttl_ms: int, + ) -> bool: ... + def get_with_ttl(self, name: str) -> tuple[bytes | None, int]: ... + def get_or_create(self, name: str, value: str, ttl: int) -> bytes: ... + + +def metadata_scope( + secret: str, namespace: str, connection_uuid: str, configuration: str +) -> str: + """Derive a private identity from trusted deployment, tenant and connection data.""" + if not secret or not namespace or not connection_uuid: + raise MetadataRefreshError("configuration") + try: + canonical: str = json.dumps( + [namespace, connection_uuid, json.loads(configuration)], + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + except (TypeError, ValueError): + raise MetadataRefreshError("configuration") from None + return hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).hexdigest() + + +@dataclass(frozen=True) +class StoredCatalog: + """Internal envelope; publication bookkeeping is never a provider revision.""" + + snapshot: CatalogSnapshot + digest: str = field(repr=False) + attempt: str = field(repr=False) + created_at: str + + +class ScopedMetadataStore: + """One shared observation with a request-wide budget and no local fallback.""" + + def __init__( + self, + backend: PublicationBackend, + scope: str, + *, + deadline: float, + before_publish: Callable[[], None] | None = None, + clock: Callable[[], float] = time.monotonic, + wait: Callable[[float], None] = time.sleep, + ) -> None: + if not math.isfinite(deadline) or not scope or "{" in scope or "}" in scope: + raise MetadataRefreshError("configuration") + self._backend: PublicationBackend = backend + self._scope: str = scope + self._deadline: float = deadline + self._lease_key: str = f"semantic-metadata:{{{scope}}}:lease" + self._snapshot_key: str = f"semantic-metadata:{{{scope}}}:snapshot" + self._generation_key: str = f"semantic-metadata:{{{scope}}}:compatibility" + self._before_publish: Callable[[], None] | None = before_publish + self._clock: Callable[[], float] = clock + self._wait: Callable[[float], None] = wait + self._observations: dict[str, str] = {} + + def _remaining(self) -> float: + return remaining_budget(self._deadline, now=self._clock()) + + def _decode(self, raw: bytes | None) -> StoredCatalog | None: + if raw is None or len(raw) > MAX_CATALOG_BYTES: + return None + try: + envelope: object = json.loads(raw) + if ( + not isinstance(envelope, dict) + or envelope.get("version") != SNAPSHOT_FORMAT_VERSION + ): + return None + if any( + not isinstance(envelope.get(key), str) + for key in ( + "payload", + "cache_token", + "observed_at", + "digest", + "attempt", + "created_at", + ) + ): + return None + if any( + not envelope[key] + for key in ( + "cache_token", + "observed_at", + "digest", + "attempt", + "created_at", + ) + ) or not envelope["cache_token"].startswith(f"{self._scope}:"): + return None + payload: str = envelope["payload"] + json.loads(payload, use_decimal=True) + if hashlib.sha256(payload.encode()).hexdigest() != envelope["digest"]: + return None + return StoredCatalog( + CatalogSnapshot( + payload, envelope["cache_token"], envelope["observed_at"] + ), + envelope["digest"], + envelope["attempt"], + envelope["created_at"], + ) + except (ValueError, UnicodeError, RecursionError): + return None + + def _load(self) -> StoredCatalog | None: + self._remaining() + stored: StoredCatalog | None = self._decode( + self._backend.get(self._snapshot_key) + ) + self._remaining() + return stored + + def _remember(self, snapshot: CatalogSnapshot) -> CatalogSnapshot: + self._observations[snapshot.cache_token] = snapshot.observed_at + return snapshot + + def observed_at(self, token: str) -> str | None: + """Read the timestamp captured with a provider's token, without backend I/O.""" + return self._observations.get(token) + + def peek(self) -> CatalogSnapshot | None: + """Read the current observation without acquiring, filling or renewing it.""" + try: + stored: StoredCatalog | None = self._load() + except RedisError: + raise MetadataRefreshError("unavailable") from None + return stored.snapshot if stored is not None else None + + def _for_deadline(self, deadline: float) -> ScopedMetadataStore: + """Narrow one call without mutating the operation or another call's budget.""" + remaining_budget(deadline, now=self._clock()) + if deadline > self._deadline: + raise MetadataRefreshError("deadline") + scoped: ScopedMetadataStore = ScopedMetadataStore( + self._backend.with_deadline(deadline), + self._scope, + deadline=deadline, + before_publish=self._before_publish, + clock=self._clock, + wait=self._wait, + ) + scoped._observations = self._observations + return scoped + + def read(self, fetch: CatalogLoader, *, deadline: float) -> CatalogSnapshot: + """Honor the explicit caller budget, including cache hits and transport.""" + return self._for_deadline(deadline)._read(fetch) + + def refresh( + self, fetch: CatalogLoader, *, deadline: float + ) -> MetadataRefreshResult: + """Publish within the caller budget, which cannot extend the host operation.""" + return self._for_deadline(deadline)._refresh(fetch) + + def _read(self, fetch: CatalogLoader) -> CatalogSnapshot: + """Wait for an owner or acquire once using the same remaining request budget.""" + try: + while True: + current: StoredCatalog | None = self._load() + if current is not None: + return self._remember(current.snapshot) + attempt: str = uuid4().hex + self._remaining() + if self._backend.set( + self._lease_key, + attempt, + px=max( + 1, + math.ceil(min(REFRESH_LEASE_SECONDS, self._remaining()) * 1000), + ), + nx=True, + ): + try: + # Another owner may have published between our read and SET NX. + current = self._load() + if current is not None: + return self._remember(current.snapshot) + return self._acquire(fetch, attempt).snapshot + finally: + self._release(attempt) + self._wait(min(READER_POLL_SECONDS, self._remaining())) + except RedisError: + self._remaining() + raise MetadataRefreshError("unavailable") from None + + def _refresh(self, fetch: CatalogLoader) -> MetadataRefreshResult: + """Publish a new observation, or report explicit contention without retry.""" + self._remaining() + attempt: str = uuid4().hex + try: + if not self._backend.set( + self._lease_key, + attempt, + px=max( + 1, math.ceil(min(REFRESH_LEASE_SECONDS, self._remaining()) * 1000) + ), + nx=True, + ): + raise MetadataRefreshError("in_progress") + try: + return self._acquire(fetch, attempt) + finally: + self._release(attempt) + except RedisError: + self._remaining() + raise MetadataRefreshError("unavailable") from None + + def _acquire(self, fetch: CatalogLoader, attempt: str) -> MetadataRefreshResult: + started: float = self._clock() + previous: StoredCatalog | None = self._load() + self._remaining() + try: + payload: str = fetch(self._deadline) + except MetadataRefreshError: + raise + except Exception: # pylint: disable=broad-except + # Provider failures cannot transport vendor payloads into host errors. + raise MetadataRefreshError("upstream") from None + self._remaining() + if not isinstance(payload, str): + raise MetadataRefreshError("invalid_payload") + try: + if len(payload.encode()) > MAX_CATALOG_BYTES: + raise MetadataRefreshError("invalid_payload") + payload = json.dumps( + json.loads(payload, use_decimal=True), + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + except (TypeError, ValueError, UnicodeError, RecursionError): + raise MetadataRefreshError("invalid_payload") from None Review Comment: A provider payload such as `{"value":1e9999999999999999999}` raises `decimal.InvalidOperation`, escaping this handler instead of becoming `invalid_payload`; the same exception also escapes `_decode()` on a stored envelope. Could both parsing boundaries handle decimal conversion failures so bad metadata cannot bypass the categorized-error/cache-miss paths? -- 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]
