feat(superset): add support for openlineage

CHANGELOG.md

@@ -4,11 +4,16 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- superset: Add the `superset-openlineage` integration that emits OpenLineage events for SQL Lab queries and chart/dashboard data fetches; config-only, inert unless enabled in `superset_config.py` ([#XXXX]).
+
 ### Removed
 
 - omid: remove 1.1.2 ([#1593]).
 
 [#1593]: https://github.com/stackabletech/docker-images/pull/1593
+[#XXXX]: https://github.com/stackabletech/docker-images/pull/XXXX
 
 ## [26.7.0] - 2026-07-21
 

superset/Dockerfile

@@ -77,6 +77,7 @@ RUN microdnf update \
 COPY superset/stackable/constraints/${PRODUCT_VERSION}/constraints.txt /tmp/constraints.txt
 COPY superset/stackable/constraints/${PRODUCT_VERSION}/build-constraints.txt /tmp/build-constraints.txt
 COPY --from=opa-authorizer-builder /tmp/opa-authorizer/dist/opa_authorizer-0.1.0-py3-none-any.whl /tmp/
+COPY superset/stackable/superset-openlineage /tmp/superset-openlineage
 
 COPY --chown=${STACKABLE_USER_UID}:0 superset/stackable/patches/patchable.toml /stackable/src/superset/stackable/patches/patchable.toml
 COPY --chown=${STACKABLE_USER_UID}:0 superset/stackable/patches/${PRODUCT_VERSION} /stackable/src/superset/stackable/patches/${PRODUCT_VERSION}
@@ -169,6 +170,13 @@ fi
 
 uv pip install --no-cache-dir /tmp/opa_authorizer-0.1.0-py3-none-any.whl
 
+# Install the Stackable OpenLineage integration for Superset. It emits OpenLineage
+# events for SQL Lab queries and chart/dashboard data fetches. It is config-only and
+# stays inert unless enabled in superset_config.py (QUERY_LOGGER, EVENT_LOGGER and
+# init_app via FLASK_APP_MUTATOR); see superset/stackable/superset-openlineage/README.md.
+# sqlglot and SQLAlchemy are provided by Superset itself; only openlineage-python is pulled in.
+uv pip install --no-cache-dir /tmp/superset-openlineage
+
 # Setuptools 82+ removed pkg_resources, which is still needed by Superset 4.x
 # dependencies. Re-pin after all other installs in case newer versions
 # have been pulled in by other dependencies.

superset/stackable/superset-openlineage/README.md

@@ -0,0 +1,76 @@
+# superset-openlineage
+
+OpenLineage integration for Apache Superset, shipped in the Stackable Superset
+image. It emits OpenLineage events when SQL Lab queries run and when dashboard
+charts fetch data.
+
+It is **config-only**: no changes are made to the Superset source tree, and the
+plugin stays completely inert unless it is explicitly enabled in
+`superset_config.py`. The package is installed into the image (see the
+`superset/Dockerfile` build stage), but activation is left to the operator/admin.
+
+## Enable it
+
+Add the following to `superset_config.py`:
+
+```python
+from superset_openlineage import (
+    build_query_logger,
+    OpenLineageEventLogger,
+    init_app,
+)
+
+QUERY_LOGGER = build_query_logger()
+EVENT_LOGGER = OpenLineageEventLogger()
+
+# Registers a teardown_request that clears the request-scoped stash
+# (required to avoid cross-request lineage mis-attribution on reused workers).
+def FLASK_APP_MUTATOR(app):
+    init_app(app)
+```
+
+## Configure the transport
+
+Point the client at a lineage backend with standard OpenLineage configuration
+(environment variables or `openlineage.yml`):
+
+```bash
+export OPENLINEAGE_URL=http://marquez:5000
+export OPENLINEAGE_API_KEY=<token>        # optional
+# or set OPENLINEAGE_DISABLED=true to turn emission off entirely
+```
+
+## Plugin settings
+
+| Env var | Default | Meaning |
+|---|---|---|
+| `SUPERSET_OPENLINEAGE_ENABLED` | `true` | Master switch for the plugin |
+| `SUPERSET_OPENLINEAGE_NAMESPACE` | `superset` | OpenLineage job namespace |
+| `SUPERSET_OPENLINEAGE_PRODUCER` | Superset repo URL | `producer` URI on events |
+| `SUPERSET_OPENLINEAGE_JOB_NAME` | `$source.$identity` | Job-name template (`string.Template`) |
+
+### Job-name template
+
+A Python `string.Template`. Placeholders: `$source`, `$identity`,
+`$sql_editor_id`, `$client_id`, `$slice_id`, `$dashboard_id`, `$sql_hash`,
+`$schema`, `$catalog`, `$database`, `$username`. Placeholders that do not apply
+in a given context render empty and surrounding separators collapse; an empty
+result falls back to `$source.$sql_hash`.
+
+## Known limitations
+
+- Failed queries and cache hits emit no events (the config-only hooks provide no
+  failure or cache signal).
+- Column-level lineage is emitted only for `CTAS`/`CVAS` statements (which have a
+  real output table); plain `SELECT` queries (charts/dashboards and most SQL Lab)
+  carry table-level inputs only.
+- Queries executed via Superset's newer SQL execution engine
+  (`superset/sql/execution/*`) are not covered yet — they invoke `QUERY_LOGGER`
+  but emit no completion event, so they produce no lineage (their pending stash
+  entries are reclaimed by the teardown hook).
+
+## Tests
+
+Unit tests live under `superset_openlineage/tests/` and require
+`openlineage-python`, `sqlglot`, `SQLAlchemy` and `pytest`. They are not
+installed into the image.

superset/stackable/superset-openlineage/pyproject.toml

@@ -0,0 +1,21 @@
+[build-system]
+requires = ["setuptools>=64"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "superset-openlineage"
+version = "0.1.0"
+description = "OpenLineage integration for Apache Superset"
+# Keep in sync with the python-version build arguments in superset/boil-config.toml.
+requires-python = ">=3.10"
+# Only openlineage-python is pulled in here. sqlglot and SQLAlchemy are runtime
+# dependencies of Apache Superset itself, so they are intentionally NOT declared
+# to avoid conflicting with Superset's own pins.
+dependencies = [
+    "openlineage-python>=1.30,<2",
+]
+
+[tool.setuptools]
+# Only the package is installed; the tests/ subpackage is intentionally excluded
+# from the wheel.
+packages = ["superset_openlineage"]

superset/stackable/superset-openlineage/superset_openlineage/__init__.py

@@ -0,0 +1,20 @@
+"""OpenLineage integration for Apache Superset (config-only plugin)."""
+
+__all__ = ["build_query_logger", "OpenLineageEventLogger", "init_app"]
+
+
+# lazy re-export to avoid import cost at config load
+def __getattr__(name: str) -> object:
+    if name == "build_query_logger":
+        from superset_openlineage.query_logger import build_query_logger
+
+        return build_query_logger
+    if name == "OpenLineageEventLogger":
+        from superset_openlineage.event_logger import OpenLineageEventLogger
+
+        return OpenLineageEventLogger
+    if name == "init_app":
+        from superset_openlineage.lifecycle import init_app
+
+        return init_app
+    raise AttributeError(name)

superset/stackable/superset-openlineage/superset_openlineage/client.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from superset_openlineage import settings
+
+logger = logging.getLogger(__name__)
+
+_client: Any = None
+_initialized: bool = False
+
+
+def reset_client() -> None:
+    global _client, _initialized
+    _client = None
+    _initialized = False
+
+
+def get_client() -> Any:
+    global _client, _initialized
+    if _initialized:
+        return _client
+    _initialized = True
+    if not settings.is_enabled():
+        _client = None
+        return None
+    try:
+        from openlineage.client import OpenLineageClient
+
+        _client = OpenLineageClient()
+    except Exception as ex:  # noqa: BLE001 - never break Superset on OL init
+        logger.warning("OpenLineage: client init failed, disabling: %s", ex)
+        _client = None
+    return _client
+
+
+def emit(event: Any) -> None:
+    client = get_client()
+    if client is None:
+        return
+    try:
+        client.emit(event)
+    except Exception as ex:  # noqa: BLE001 - emission must never break a query
+        logger.warning("OpenLineage: emit failed, dropping event: %s", ex)

superset/stackable/superset-openlineage/superset_openlineage/context.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from typing import Any
+
+_PENDING: ContextVar[list["PendingRun"] | None] = ContextVar("ol_pending", default=None)
+
+
+@dataclass
+class PendingRun:
+    run_id: str
+    sql: str
+    dialect: str | None
+    source: str
+    sql_hash: str
+    schema: str | None
+    catalog: str | None
+    database: str | None
+    username: str | None
+    sql_editor_id: str | None
+    client_id: str | None
+    start_time: float
+    inputs: list[Any] = field(default_factory=list)
+    outputs: list[Any] = field(default_factory=list)
+    column_lineage: Any = None
+
+
+def _stack() -> list[PendingRun]:
+    stack = _PENDING.get()
+    if stack is None:
+        stack = []
+        _PENDING.set(stack)
+    return stack
+
+
+def push_pending(run: PendingRun) -> None:
+    stack = list(_stack())
+    stack.append(run)
+    _PENDING.set(stack)
+
+
+def pop_pending() -> PendingRun | None:
+    stack = list(_stack())
+    if not stack:
+        return None
+    run = stack.pop()
+    _PENDING.set(stack)
+    return run
+
+
+def clear_pending() -> None:
+    _PENDING.set([])

superset/stackable/superset-openlineage/superset_openlineage/event_logger.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from openlineage.client.event_v2 import Job, Run, RunEvent, RunState
+from openlineage.client.facet_v2 import job_type_job, sql_job
+
+from superset.utils.log import DBEventLogger
+from superset_openlineage import settings
+from superset_openlineage.client import emit
+from superset_openlineage.context import PendingRun, pop_pending
+from superset_openlineage.facets import SupersetJobFacet
+from superset_openlineage.jobname import render_job_name, resolve_identity
+
+logger = logging.getLogger(__name__)
+
+
+class OpenLineageEventLogger(DBEventLogger):
+    """DBEventLogger that also emits OpenLineage run events on query success."""
+
+    def log(  # type: ignore[override]
+        self, user_id: int | None, action: str, *args: Any, **kwargs: Any
+    ) -> None:
+        # Always preserve default DB audit logging.
+        self._persist(user_id, action, *args, **kwargs)
+        if action != "execute_sql":
+            return
+        try:
+            self._emit_lineage(kwargs)
+        except Exception as ex:  # noqa: BLE001 - emission must never break a query
+            logger.warning("OpenLineage: event emission failed: %s", ex)
+
+    def _persist(
+        self, user_id: int | None, action: str, *args: Any, **kwargs: Any
+    ) -> None:
+        super().log(user_id, action, *args, **kwargs)
+
+    def _emit_lineage(self, kwargs: dict[str, Any]) -> None:
+        pending = pop_pending()
+        if pending is None:
+            return
+        job = self._build_job(pending, kwargs)
+        run = Run(runId=pending.run_id)
+        producer = settings.get_producer()
+        start_iso = (
+            datetime.fromtimestamp(pending.start_time, tz=timezone.utc).isoformat()
+            if pending.start_time
+            else datetime.now(timezone.utc).isoformat()
+        )
+        end_iso = datetime.now(timezone.utc).isoformat()
+        emit(
+            RunEvent(
+                eventType=RunState.START,
+                eventTime=start_iso,
+                run=run,
+                job=job,
+                inputs=pending.inputs,
+                outputs=pending.outputs,
+                producer=producer,
+            )
+        )
+        emit(
+            RunEvent(
+                eventType=RunState.COMPLETE,
+                eventTime=end_iso,
+                run=run,
+                job=job,
+                inputs=pending.inputs,
+                outputs=pending.outputs,
+                producer=producer,
+            )
+        )
+
+    def _build_job(self, pending: PendingRun, kwargs: dict[str, Any]) -> Job:
+        values = {
+            "source": pending.source,
+            "sql_editor_id": pending.sql_editor_id,
+            "client_id": pending.client_id,
+            "slice_id": kwargs.get("slice_id"),
+            "dashboard_id": kwargs.get("dashboard_id"),
+            "sql_hash": pending.sql_hash,
+            "schema": pending.schema,
+            "catalog": pending.catalog,
+            "database": pending.database,
+            "username": pending.username,
+        }
+        values["identity"] = resolve_identity(values)
+        name = render_job_name(settings.get_job_name_template(), values)
+        job_facets: dict[str, Any] = {
+            "jobType": job_type_job.JobTypeJobFacet(
+                processingType="BATCH",
+                integration="SUPERSET",
+                jobType="QUERY",
+            ),
+            "sql": sql_job.SQLJobFacet(
+                query=pending.sql, dialect=pending.dialect or ""
+            ),
+            "superset": SupersetJobFacet(
+                source=pending.source,
+                dashboardId=kwargs.get("dashboard_id"),
+                username=pending.username,
+            ),
+        }
+        return Job(namespace=settings.get_namespace(), name=name, facets=job_facets)

superset/stackable/superset-openlineage/superset_openlineage/facets.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+import attr
+from openlineage.client.facet_v2 import JobFacet
+
+
+@attr.define
+class SupersetJobFacet(JobFacet):
+    source: str | None = None
+    dashboardId: int | None = None  # noqa: N815 - OpenLineage uses camelCase
+    username: str | None = None