fix(lineage): redact chart dataset details when caller lacks access

Mirrors the dashboard lineage endpoint's gating: schema/table/database name are only exposed when the caller can access the underlying datasource, so the chart endpoint no longer leaks that info unconditionally. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
fix(lineage): nullable schema fields, lazy modal fetch, N+1 + type hints
2026-08-20 07:01:17 +00:00 · 2026-08-19 21:25:38 -07:00 · 2026-08-19 21:18:46 -07:00 · 2026-08-19 21:18:21 -07:00 · 2026-08-19 21:18:21 -07:00 · 2026-08-19 21:18:21 -07:00
80 changed files with 4675 additions and 145 deletions
@@ -48,7 +48,7 @@ jobs:
          python-version: "3.11"

      - name: Install uv
-        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        uses: astral-sh/setup-uv@ae62891fec2bb8e7d6c99fc78c9fec3a63790f8d # v10.0.0
        with:
          python-version: "3.11"
          enable-cache: true
@@ -64,8 +64,8 @@
    "antd": "^6.6.0",
    "baseline-browser-mapping": "^2.11.13",
    "caniuse-lite": "^1.0.30001809",
-    "docusaurus-plugin-openapi-docs": "^5.1.3",
-    "docusaurus-theme-openapi-docs": "^5.1.3",
+    "docusaurus-plugin-openapi-docs": "^5.2.0",
+    "docusaurus-theme-openapi-docs": "^5.2.0",
    "js-yaml": "^5.2.3",
    "json-bigint": "^1.0.0",
    "prism-react-renderer": "^2.4.1",
@@ -78,7 +78,7 @@
    "remark-import-partial": "^0.0.2",
    "reselect": "^5.2.0",
    "storybook": "^10.5.7",
-    "swagger-ui-react": "^5.32.12",
+    "swagger-ui-react": "^5.32.13",
    "swc-loader": "^0.2.7",
    "tinycolor2": "^1.4.2",
    "unist-util-visit": "^5.1.0"
@@ -93,7 +93,7 @@
    "@typescript-eslint/parser": "^8.67.0",
    "eslint": "^9.39.2",
    "eslint-plugin-react": "^7.37.5",
-    "globals": "^17.9.0",
+    "globals": "^17.10.0",
    "oxfmt": "^0.63.0",
    "typescript": "~6.0.3",
    "typescript-eslint": "^8.67.0",
@@ -8014,10 +8014,10 @@ doctrine@^2.1.0:
  dependencies:
    esutils "^2.0.2"

-docusaurus-plugin-openapi-docs@^5.1.3:
-  version "5.1.3"
-  resolved "https://registry.yarnpkg.com/docusaurus-plugin-openapi-docs/-/docusaurus-plugin-openapi-docs-5.1.3.tgz#b8cd5f8451aaf881deb1a744a8295685f1681865"
-  integrity sha512-HnpblSBdXoR39VNTIW9zWERUsMJxXOpvdQoBKyaTkUBPwCM48Z76+ndo2yO2vADq+EhWjJlfxL1DUzCrgNjThQ==
+docusaurus-plugin-openapi-docs@^5.2.0:
+  version "5.2.0"
+  resolved "https://registry.yarnpkg.com/docusaurus-plugin-openapi-docs/-/docusaurus-plugin-openapi-docs-5.2.0.tgz#8318ec90cd21fed023be57696211af7d72fd81db"
+  integrity sha512-MjrfRAMB64uvdxRVz6L9AXWe4QFjCdoBAzYs306yyI3nnXHsFj2lv2FnLA90JV9CAUZaGiYMvvkzBo2Nrkq/9w==
  dependencies:
    "@apidevtools/json-schema-ref-parser" "^15.3.3"
    "@redocly/openapi-core" "^2.25.2"
@@ -8035,10 +8035,10 @@ docusaurus-plugin-openapi-docs@^5.1.3:
    swagger2openapi "^7.0.8"
    xml-formatter "^3.6.6"

-docusaurus-theme-openapi-docs@^5.1.3:
-  version "5.1.3"
-  resolved "https://registry.yarnpkg.com/docusaurus-theme-openapi-docs/-/docusaurus-theme-openapi-docs-5.1.3.tgz#e23644a63785352abbc76e42760c0dfdff3669e1"
-  integrity sha512-npbD1QahtjAEmrOet/86i5fTmcJX4/rPhVT+c0qKjm7StUNbyqjwchSVBQuU1rB69T51JOA9TpT/y6QcB9Xjvw==
+docusaurus-theme-openapi-docs@^5.2.0:
+  version "5.2.0"
+  resolved "https://registry.yarnpkg.com/docusaurus-theme-openapi-docs/-/docusaurus-theme-openapi-docs-5.2.0.tgz#6d93a74e2e3cf0ae77d24e1c4144bd2e74a52115"
+  integrity sha512-L0b80LzaMUfr76a9EQXRPCf8nxkEz8Xo6Aknnke1UeE2oXsgoiVki6U+RTE7GmJRjO8zSNKXyckGmGmqqWuHeA==
  dependencies:
    "@hookform/error-message" "^2.0.1"
    "@reduxjs/toolkit" "^2.8.2"
@@ -8123,7 +8123,7 @@ domhandler@^5.0.2, domhandler@^5.0.3:
  dependencies:
    domelementtype "^2.3.0"

-dompurify@^3.3.3, dompurify@^3.4.12:
+dompurify@^3.3.3, dompurify@^3.4.13:
  version "3.4.13"
  resolved "https://registry.yarnpkg.com/dompurify/-/dompurify-3.4.13.tgz#fc28949d59f92d62e28a3a764bcbeee35897a1be"
  integrity sha512-2vmYIoqjze2d+kakP8S/nS5shfsl587kzwEjcGlTdiksUVgFHnFCsLYDVj/JNqJVOQZGSYBTmuycv0PodwmnMQ==
@@ -9174,10 +9174,10 @@ globals@^14.0.0:
  resolved "https://registry.yarnpkg.com/globals/-/globals-14.0.0.tgz#898d7413c29babcf6bafe56fcadded858ada724e"
  integrity sha512-oahGvuMGQlPw/ivIYBjVSrWAfWLBeku5tpPE2fOPLi+WHffIWbuh2tCjhyQhTBPMf5E9jDEH4FOmTYgYwbKwtQ==

-globals@^17.9.0:
-  version "17.9.0"
-  resolved "https://registry.yarnpkg.com/globals/-/globals-17.9.0.tgz#e43f252d6bbe71508da43902a1709c8895a59f70"
-  integrity sha512-m/MvAW61QVU5VDNF1Vj8axt016h8w7L5TU1e9zlab7XIttAT2YAlCwl75K1fOqvMM9apmD7lbCIRhpfkhmxhCg==
+globals@^17.10.0:
+  version "17.10.0"
+  resolved "https://registry.yarnpkg.com/globals/-/globals-17.10.0.tgz#f9dbd847ae99e236f98b13095e2426ac3b25a45c"
+  integrity sha512-V0kztuWST2k8A/VbxAY8+L+7+Rgo3fyA24IHRLrZp7HOzJjV0gHSaZUjK9lpP/IrBSNite2tZ1prhRkinRu1CA==

 globalthis@^1.0.4:
  version "1.0.4"
@@ -10284,10 +10284,10 @@ js-levenshtein@^1.1.6:
  resolved "https://registry.yarnpkg.com/js-tokens/-/js-tokens-4.0.0.tgz#19203fb59991df98e3a287050d4647cdeaf32499"
  integrity sha512-RdJUflcE3cUzKiMqQgsCu06FPu9UdIJO0beYbPhHN4k6apgJtifcoCtT9bcxOpYBtpD2kCM6Sbzg4CausW/PKQ==

-js-yaml@4.1.0, js-yaml@=4.3.0, js-yaml@^4.1.0, js-yaml@^4.1.1, js-yaml@^4.2.0, js-yaml@^4.3.0:
-  version "4.3.0"
-  resolved "https://registry.yarnpkg.com/js-yaml/-/js-yaml-4.3.0.tgz#d1900572a7f7cf0b5f540c83673e60bad3436592"
-  integrity sha512-1td788aAnnZ5qs7V2QIRl1owjtYpbKt749Y3xauqQgwIIGF/xXWz1wMTEBx5O3LK3lXLVuqXPdPxj2BoFHaW9Q==
+js-yaml@4.1.0, js-yaml@=4.3.1, js-yaml@^4.1.0, js-yaml@^4.1.1, js-yaml@^4.2.0, js-yaml@^4.3.0:
+  version "4.3.1"
+  resolved "https://registry.yarnpkg.com/js-yaml/-/js-yaml-4.3.1.tgz#01216c001d67f48e2cd560d708c7af21090a3848"
+  integrity sha512-CY6crGq313MX8GkwvB7tzgp99vjQxY1++5y10/BKN/GUfHqWaOGQMNZkBvqSzsZKWk/ijwHlWzzkLulsGHhjWQ==
  dependencies:
    argparse "^2.0.1"

@@ -15103,10 +15103,10 @@ swagger-client@^3.37.8:
    "@swagger-api/apidom-parser-adapter-openapi-yaml-3-2" "^1.12.0"
    "@swagger-api/apidom-parser-adapter-yaml-1-2" "^1.12.0"

-swagger-ui-react@^5.32.12:
-  version "5.32.12"
-  resolved "https://registry.yarnpkg.com/swagger-ui-react/-/swagger-ui-react-5.32.12.tgz#47525a26774eb02db0e6203af72f5b32fa6205cc"
-  integrity sha512-WCdkNOQyMTZDu+z356FpwVWHf1dwZgQPUjdQPh1L4r7jULaJTKKlIItXq6WsZdYeXvsHndMdxxccEQXOAroUHQ==
+swagger-ui-react@^5.32.13:
+  version "5.32.13"
+  resolved "https://registry.yarnpkg.com/swagger-ui-react/-/swagger-ui-react-5.32.13.tgz#04c96140b0a2d4ea01ebec4d4cfc655d5ed9a500"
+  integrity sha512-XIDl+Ny6kE1N8wpSPiOFrjPfAevs4GR4XmV6BT6NLMikkMFIbIVocWbA8pnKYyYXQe8Rccfli5o2zDfySw0FnQ==
  dependencies:
    "@babel/runtime-corejs3" "^7.27.1"
    "@scarf/scarf" "=1.4.0"
@@ -15115,11 +15115,11 @@ swagger-ui-react@^5.32.12:
    classnames "^2.5.1"
    css.escape "1.5.1"
    deep-extend "0.6.0"
-    dompurify "^3.4.12"
+    dompurify "^3.4.13"
    ieee754 "^1.2.1"
    immutable "^4.3.9"
    js-file-download "^0.4.12"
-    js-yaml "=4.3.0"
+    js-yaml "=4.3.1"
    lodash "^4.18.1"
    prop-types "^15.8.1"
    randexp "^0.5.3"
@@ -186,7 +186,7 @@
        "@storybook/test-runner": "0.24.4",
        "@svgr/webpack": "^8.1.0",
        "@swc/core": "^1.15.47",
-        "@swc/plugin-emotion": "^14.15.0",
+        "@swc/plugin-emotion": "^14.19.0",
        "@swc/plugin-transform-imports": "^12.5.0",
        "@testing-library/dom": "^10.4.1",
        "@testing-library/jest-dom": "^7.0.1",
@@ -11808,9 +11808,9 @@
      }
    },
    "node_modules/@swc/plugin-emotion": {
-      "version": "14.15.0",
-      "resolved": "https://registry.npmjs.org/@swc/plugin-emotion/-/plugin-emotion-14.15.0.tgz",
-      "integrity": "sha512-nCsTO7mOOPz2UnT3N6YWb014uI0CVxeKg53A/KM/CvuSIE6H3KPkhaziJQ3q2jI3u3LfFuDKEnU5ZmB1330Dqg==",
+      "version": "14.19.0",
+      "resolved": "https://registry.npmjs.org/@swc/plugin-emotion/-/plugin-emotion-14.19.0.tgz",
+      "integrity": "sha512-0/q84ro0a7kdjpYpn9Wmi5/RLHYuSwYjO638lE5ZBQfIvYpSLJxbEgLsObCmdH4KPe2stoN8plVKUpCsKPggaw==",
      "dev": true,
      "license": "Apache-2.0",
      "dependencies": {
@@ -20612,7 +20612,7 @@
      "version": "0.8.0",
      "resolved": "https://registry.npmjs.org/expect-playwright/-/expect-playwright-0.8.0.tgz",
      "integrity": "sha512-+kn8561vHAY+dt+0gMqqj1oY+g5xWrsuGMk4QGxotT2WS545nVqqjs37z6hrYfIuucwqthzwJfCJUEYqixyljg==",
-      "deprecated": "⚠️ The 'expect-playwright' package is deprecated. The Playwright core assertions (via @playwright/test) now cover the same functionality. Please migrate to built-in expect. See https://playwright.dev/docs/test-assertions for migration.",
+      "deprecated": "\u26a0\ufe0f The 'expect-playwright' package is deprecated. The Playwright core assertions (via @playwright/test) now cover the same functionality. Please migrate to built-in expect. See https://playwright.dev/docs/test-assertions for migration.",
      "dev": true,
      "license": "MIT"
    },
@@ -26023,7 +26023,7 @@
      "version": "0.4.0",
      "resolved": "https://registry.npmjs.org/jest-process-manager/-/jest-process-manager-0.4.0.tgz",
      "integrity": "sha512-80Y6snDyb0p8GG83pDxGI/kQzwVTkCxc7ep5FPe/F6JYdvRDhwr6RzRmPSP7SEwuLhxo80lBS/NqOdUIbHIfhw==",
-      "deprecated": "⚠️ The 'jest-process-manager' package is deprecated. Please migrate to Playwright's built-in test runner (@playwright/test) which now includes full Jest-style features and parallel testing. See https://playwright.dev/docs/intro for details.",
+      "deprecated": "\u26a0\ufe0f The 'jest-process-manager' package is deprecated. Please migrate to Playwright's built-in test runner (@playwright/test) which now includes full Jest-style features and parallel testing. See https://playwright.dev/docs/intro for details.",
      "dev": true,
      "license": "MIT",
      "dependencies": {
@@ -43073,6 +43073,15 @@
        "node": ">=12"
      }
    },
+    "packages/superset-ui-core/node_modules/dompurify": {
+      "version": "3.4.13",
+      "resolved": "https://registry.npmjs.org/dompurify/-/dompurify-3.4.13.tgz",
+      "integrity": "sha512-2vmYIoqjze2d+kakP8S/nS5shfsl587kzwEjcGlTdiksUVgFHnFCsLYDVj/JNqJVOQZGSYBTmuycv0PodwmnMQ==",
+      "license": "(MPL-2.0 OR Apache-2.0)",
+      "optionalDependencies": {
+        "@types/trusted-types": "^2.0.7"
+      }
+    },
    "packages/superset-ui-core/node_modules/react-ace": {
      "version": "14.0.1",
      "resolved": "https://registry.npmjs.org/react-ace/-/react-ace-14.0.1.tgz",
@@ -43420,22 +43429,13 @@
      "license": "Apache-2.0",
      "dependencies": {
        "d3": "^3.5.17",
-        "prop-types": "^15.8.1",
-        "react": "^19.2.7"
+        "prop-types": "^15.8.1"
      },
      "peerDependencies": {
        "@apache-superset/core": "*",
        "@superset-ui/chart-controls": "*",
-        "@superset-ui/core": "*"
-      }
-    },
-    "plugins/plugin-chart-chord/node_modules/react": {
-      "version": "19.2.8",
-      "resolved": "https://registry.npmjs.org/react/-/react-19.2.8.tgz",
-      "integrity": "sha512-PWaYA1L/q9u2u7xYQi+Y3L3Yfnie7XyLeaJICV1MGD6LprsBxcAqGjYyr0eY3p+QdsA+x/Irkt4Qif8D63+Sbw==",
-      "license": "MIT",
-      "engines": {
-        "node": ">=0.10.0"
+        "@superset-ui/core": "*",
+        "react": "^18.3.0"
      }
    },
    "plugins/plugin-chart-country-map": {
@@ -263,7 +263,7 @@
    "@storybook/test-runner": "0.24.4",
    "@svgr/webpack": "^8.1.0",
    "@swc/core": "^1.15.47",
-    "@swc/plugin-emotion": "^14.15.0",
+    "@swc/plugin-emotion": "^14.19.0",
    "@swc/plugin-transform-imports": "^12.5.0",
    "@testing-library/dom": "^10.4.1",
    "@testing-library/jest-dom": "^7.0.1",
@@ -130,6 +130,7 @@ export enum GenericDataType {
  String = 1,
  Temporal = 2,
  Boolean = 3,
+  MultiValue = 4,
 }

 /**
@@ -28,6 +28,7 @@ import {
  FieldBinaryOutlined,
  FieldStringOutlined,
  NumberOutlined,
+  UnorderedListOutlined,
 } from '@ant-design/icons';
 import { Icons } from '@superset-ui/core/components';

@@ -72,6 +73,10 @@ export function ColumnTypeLabel({ type }: ColumnTypeLabelProps) {
    typeIcon = <FieldBinaryOutlined aria-label={t('boolean type icon')} />;
  } else if (type === GenericDataType.Temporal) {
    typeIcon = <ClockCircleOutlined aria-label={t('temporal type icon')} />;
+  } else if (type === GenericDataType.MultiValue) {
+    typeIcon = (
+      <UnorderedListOutlined aria-label={t('multi-value type icon')} />
+    );
  }

  return <TypeIconWrapper>{typeIcon}</TypeIconWrapper>;
@@ -64,4 +64,21 @@ describe('ColumnOption', () => {
    renderColumnTypeLabel({ type: GenericDataType.Temporal });
    expect(screen.getByLabelText('temporal type icon')).toBeVisible();
  });
+  test('multi-value (array) type shows list icon', () => {
+    renderColumnTypeLabel({ type: GenericDataType.MultiValue });
+    expect(screen.getByLabelText('multi-value type icon')).toBeVisible();
+  });
+});
+
+describe('GenericDataType enum parity', () => {
+  // These numeric values are shared with the backend enum in
+  // superset/utils/core.py (GenericDataType). They must stay in sync because
+  // the backend serializes columns using these integers.
+  test('values match the backend contract', () => {
+    expect(GenericDataType.Numeric).toBe(0);
+    expect(GenericDataType.String).toBe(1);
+    expect(GenericDataType.Temporal).toBe(2);
+    expect(GenericDataType.Boolean).toBe(3);
+    expect(GenericDataType.MultiValue).toBe(4);
+  });
 });
@@ -47,6 +47,10 @@ export default defineConfig({
  // Retry logic - 2 retries in CI, 0 locally
  retries: process.env.CI ? 2 : 0,

+  // Disable capturing Git commit info as the project's history is increasingly dense
+  // and breach Playwright's default 3-seconds `git` command timeout limit
+  captureGitInfo: { commit: false, diff: false },
+
  // Reporter configuration - multiple reporters for better visibility
  reporter: process.env.CI
    ? [
@@ -30,12 +30,12 @@
  },
  "dependencies": {
    "d3": "^3.5.17",
-    "prop-types": "^15.8.1",
-    "react": "^19.2.7"
+    "prop-types": "^15.8.1"
  },
  "peerDependencies": {
    "@apache-superset/core": "*",
    "@superset-ui/chart-controls": "*",
-    "@superset-ui/core": "*"
+    "@superset-ui/core": "*",
+    "react": "^18.3.0"
  }
 }
@@ -17,6 +17,10 @@
 * under the License.
 */

+import { getNumberFormatter } from '@superset-ui/core';
+import { render, fireEvent } from '../../../../spec/helpers/testing-library';
+import BigNumberVis from './BigNumberViz';
+
 /**
 * Tests for the color threshold formatter logic in BigNumberViz.
 *
@@ -83,3 +87,33 @@ describe('BigNumberViz color formatters', () => {
    expect(getColorFromValue).not.toHaveBeenCalled();
  });
 });
+
+describe('BigNumberViz context menu', () => {
+  test('invokes onContextMenu and stops the event bubbling to ancestor handlers', () => {
+    const onContextMenu = jest.fn();
+    const ancestorHandler = jest.fn();
+
+    const { container } = render(
+      <div onContextMenu={ancestorHandler}>
+        <BigNumberVis
+          width={200}
+          height={100}
+          bigNumber={42}
+          headerFormatter={getNumberFormatter()}
+          headerFontSize={0.3}
+          subheaderFontSize={0.125}
+          subtitleFontSize={0.125}
+          subtitle=""
+          refs={{}}
+          onContextMenu={onContextMenu}
+        />
+      </div>,
+    );
+
+    const headerLine = container.querySelector('.header-line');
+    fireEvent.contextMenu(headerLine!, { clientX: 10, clientY: 20 });
+
+    expect(onContextMenu).toHaveBeenCalledWith(10, 20);
+    expect(ancestorHandler).not.toHaveBeenCalled();
+  });
+});
@@ -224,6 +224,7 @@ function BigNumberVis({
    const handleContextMenu = (e: MouseEvent<HTMLDivElement>) => {
      if (onContextMenu) {
        e.preventDefault();
+        e.stopPropagation();
        onContextMenu(e.nativeEvent.clientX, e.nativeEvent.clientY);
      }
    };
@@ -390,6 +390,7 @@ export default function transformProps(chartProps: EchartsGanttChartProps) {
    [GenericDataType.String]: undefined,
    [GenericDataType.Temporal]: tooltipTimeFormatter,
    [GenericDataType.Boolean]: undefined,
+    [GenericDataType.MultiValue]: undefined,
  };

  const echartOptions: EChartsCoreOption = {
@@ -62,6 +62,7 @@ import {
  Col,
  Divider,
  EditableTitle,
+  Empty,
  Flex,
  FormLabel,
  Icons,
@@ -92,6 +93,8 @@ import { DatabaseSelector } from '../../../DatabaseSelector';
 import SpatialControl from 'src/explore/components/controls/SpatialControl';
 import CollectionTable from '../CollectionTable';
 import Fieldset from '../Fieldset';
+import { useDatasetLineage } from 'src/hooks/apiResources';
+import { LineageView } from 'src/features/lineage';
 import Field from '../Field';
 import { fetchSyncedColumns, updateColumns } from '../../utils';
 import DatasetUsageTab from './components/DatasetUsageTab';
@@ -398,6 +401,20 @@ const StyledTableTabWrapper = styled.div`
  }
 `;

+// Functional wrapper for the lineage tab, since hooks can't be used directly in
+// the DatasourceEditor class component.
+function DatasetLineageTab({ datasourceId }: { datasourceId?: number }) {
+  const lineageResource = useDatasetLineage(datasourceId ?? '');
+  if (!datasourceId) {
+    return (
+      <Empty
+        description={t('Lineage is available after the dataset is saved')}
+      />
+    );
+  }
+  return <LineageView lineageResource={lineageResource} entityType="dataset" />;
+}
+
 const DefaultColumnSettingsContainer = styled.div`
  ${({ theme }) => css`
    margin-bottom: ${theme.sizeUnit * 4}px;
@@ -450,6 +467,7 @@ const TABS_KEYS = {
  COLUMNS: 'COLUMNS',
  CALCULATED_COLUMNS: 'CALCULATED_COLUMNS',
  USAGE: 'USAGE',
+  LINEAGE: 'LINEAGE',
  FOLDERS: 'FOLDERS',
  SETTINGS: 'SETTINGS',
  SPATIAL: 'SPATIAL',
@@ -1627,9 +1645,7 @@ function DatasourceEditor({
              {t(
                'Default URL to redirect to when accessing from the dataset list page. Accepts relative URLs such as',
              )}{' '}
-              <Typography.Text code>
-                /superset/dashboard/{'{id}'}/
-              </Typography.Text>
+              <Typography.Text code>/dashboard/{'{id}'}/</Typography.Text>
            </>
          }
          control={<TextControl controlId="default_endpoint" />}
@@ -2494,6 +2510,15 @@ function DatasourceEditor({
          </StyledTableTabWrapper>
        ),
      },
+      {
+        key: TABS_KEYS.LINEAGE,
+        label: t('Lineage'),
+        children: (
+          <StyledTableTabWrapper>
+            <DatasetLineageTab datasourceId={datasource.id} />
+          </StyledTableTabWrapper>
+        ),
+      },
      ...(isFeatureEnabled(FeatureFlag.DatasetFolders)
        ? [
            {
@@ -71,6 +71,17 @@ test('renders Tabs', async () => {
  expect(screen.getByTestId('edit-dataset-tabs')).toBeInTheDocument();
 });

+test('recommends a registered client route for the default URL', async () => {
+  await asyncRender(createProps());
+
+  userEvent.click(screen.getByRole('tab', { name: 'Settings' }));
+
+  expect(await screen.findByText('/dashboard/{id}/')).toBeInTheDocument();
+  expect(
+    screen.queryByText('/superset/dashboard/{id}/'),
+  ).not.toBeInTheDocument();
+});
+
 test('can sync columns from source', async () => {
  const testProps = createProps();
  await asyncRender({
@@ -40,6 +40,7 @@ import { HeaderDropdownProps } from 'src/dashboard/components/Header/types';
 import { usePermissions } from 'src/hooks/usePermissions';
 import { openVersionHistoryPanel } from 'src/features/versionHistory/reducer';
 import getUserName from 'src/utils/getUserName';
+import { LineageModal } from 'src/features/lineage';

 export const useHeaderActionsMenu = ({
  customCss,
@@ -300,6 +301,23 @@ export const useHeaderActionsMenu = ({
      });
    }

+    // View lineage (available in both view and edit mode; lineage is
+    // read-only information about the dashboard's upstream assets)
+    if (dashboardId) {
+      menuItems.push(
+        createModalMenuItem(
+          MenuKeys.ViewLineage,
+          <LineageModal
+            entityType="dashboard"
+            entityId={dashboardId}
+            triggerNode={
+              <div data-test="view-lineage-menu-item">{t('View lineage')}</div>
+            }
+          />,
+        ),
+      );
+    }
+
    // Edit properties
    if (editMode) {
      menuItems.push({
@@ -468,7 +468,7 @@ function SliceAdder({
          <AutoSizer>
            {({ height, width }: { height: number; width: number }) => (
              <List
-                style={{ width, height }}
+                style={{ width, height, maxHeight: height }}
                rowCount={filteredSlices.length}
                rowHeight={DEFAULT_CELL_HEIGHT}
                rowProps={listRowProps}
@@ -407,4 +407,5 @@ export enum MenuKeys {
  ExportPivotXlsx = 'export_pivot_xlsx',
  EmbedCode = 'embed_code',
  VersionHistory = 'version_history',
+  ViewLineage = 'view_lineage',
 }
@@ -148,7 +148,7 @@ export const DatasourceItems = ({

  return (
    <List
-      style={{ width: width - BORDER_WIDTH, height }}
+      style={{ width: width - BORDER_WIDTH, height, maxHeight: height }}
      rowHeight={rowHeight}
      rowCount={flattenedItems.length}
      rowProps={rowProps}
@@ -251,4 +251,11 @@ export const DEFAULT_CONFIG_FORM_LAYOUT: ColumnConfigFormLayout = {
      { name: 'horizontalAlign', override: { defaultValue: 'left' } },
    ],
  ],
+  [GenericDataType.MultiValue]: [
+    [
+      'columnWidth',
+      { name: 'horizontalAlign', override: { defaultValue: 'left' } },
+    ],
+    ['truncateLongCells'],
+  ],
 };
@@ -270,6 +270,74 @@ describe('AdhocFilter', () => {
    });
    expect(adhocFilter.comparator).toBe(undefined);
  });
+  // Charts saved before #32701 persisted `==` as the operation for IS_TRUE and
+  // IS_FALSE, alongside a boolean comparator. `translateToSql` and the backend
+  // both key off `operator`, so dropping the comparator would render such a
+  // filter as `col =` and query it as `col IS NULL`.
+  test('keeps the legacy boolean comparator for IS_TRUE', () => {
+    const adhocFilter = new AdhocFilter({
+      expressionType: ExpressionTypes.Simple,
+      subject: 'col',
+      operator: '==',
+      operatorId: Operators.IsTrue,
+      comparator: true,
+      clause: Clauses.Where,
+    });
+    expect(adhocFilter.operator).toBe('==');
+    expect(adhocFilter.comparator).toBe(true);
+    expect(adhocFilter.translateToSql()).toBe("col = 'TRUE'");
+  });
+  test('keeps the legacy boolean comparator for IS_FALSE', () => {
+    const adhocFilter = new AdhocFilter({
+      expressionType: ExpressionTypes.Simple,
+      subject: 'col',
+      operator: '==',
+      operatorId: Operators.IsFalse,
+      comparator: false,
+      clause: Clauses.Where,
+    });
+    expect(adhocFilter.operator).toBe('==');
+    expect(adhocFilter.comparator).toBe(false);
+    expect(adhocFilter.translateToSql()).toBe("col = 'FALSE'");
+  });
+  test('restores the boolean even when the stored comparator is missing', () => {
+    const adhocFilter = new AdhocFilter({
+      expressionType: ExpressionTypes.Simple,
+      subject: 'col',
+      operator: '==',
+      operatorId: Operators.IsTrue,
+      clause: Clauses.Where,
+    });
+    expect(adhocFilter.comparator).toBe(true);
+  });
+  test('keeps a legacy boolean filter intact when the control re-posts it', () => {
+    const stored = {
+      expressionType: ExpressionTypes.Simple,
+      subject: 'col',
+      operator: '==',
+      operatorId: Operators.IsTrue,
+      comparator: true,
+      clause: Clauses.Where,
+    };
+    // DndFilterSelect wraps props.value and hands those instances to onChange
+    const posted = JSON.parse(JSON.stringify(new AdhocFilter(stored)));
+    expect(posted.operator).toBe('==');
+    expect(posted.comparator).toBe(true);
+    expect(posted.operatorId).toBe(Operators.IsTrue);
+  });
+  test('leaves a genuine equality filter on a boolean value alone', () => {
+    const adhocFilter = new AdhocFilter({
+      expressionType: ExpressionTypes.Simple,
+      subject: 'col',
+      operator: '==',
+      operatorId: Operators.Equals,
+      comparator: true,
+      clause: Clauses.Where,
+    });
+    expect(adhocFilter.operator).toBe('==');
+    expect(adhocFilter.comparator).toBe(true);
+    expect(adhocFilter.translateToSql()).toBe("col = 'TRUE'");
+  });
  test('sets the label properly if subject is a string', () => {
    const adhocFilter = new AdhocFilter({
      expressionType: ExpressionTypes.Simple,
@@ -30,6 +30,15 @@ const CUSTOM_OPERATIONS = [...CUSTOM_OPERATORS].map(
  op => OPERATOR_ENUM_TO_OPERATOR_TYPE[op].operation,
 );

+// Charts saved before #32701 store `==` for IS_TRUE/IS_FALSE with the boolean
+// in the comparator; blanking it makes them query `col IS NULL`. Restoring it
+// leaves the emitted SQL untouched -- reconciling `operator` to `IS TRUE`
+// would not, and Druid rejects that predicate on VARCHAR columns.
+const LEGACY_BOOLEAN_COMPARATORS = new Map<string, boolean>([
+  [Operators.IsTrue, true],
+  [Operators.IsFalse, false],
+]);
+
 interface AdhocFilterInput {
  expressionType?: string;
  subject?: string | { column_name?: string; [key: string]: unknown } | null;
@@ -77,6 +86,16 @@ export default class AdhocFilter {
      ) {
        this.comparator = undefined;
      }
+      if (
+        this.operator ===
+          OPERATOR_ENUM_TO_OPERATOR_TYPE[Operators.Equals].operation &&
+        adhocFilter.operatorId &&
+        LEGACY_BOOLEAN_COMPARATORS.has(adhocFilter.operatorId)
+      ) {
+        this.comparator = LEGACY_BOOLEAN_COMPARATORS.get(
+          adhocFilter.operatorId,
+        );
+      }
      this.clause = adhocFilter.clause || Clauses.Where;
      this.sqlExpression = null;
    } else if (this.expressionType === ExpressionTypes.Sql) {
@@ -367,8 +367,22 @@ function AdhocFilterEditPopover({
              </ErrorBoundary>
            ),
          },
-          ...(datasource?.type === 'semantic_view'
-            ? []
+          ...(datasource?.type === 'semantic_view' ||
+          [
+            Operators.ContainsAny,
+            Operators.ContainsAll,
+            Operators.IsEmpty,
+            Operators.IsNotEmpty,
+            Operators.LengthEquals,
+            Operators.LengthGreaterThan,
+            Operators.LengthLessThan,
+            Operators.LengthGreaterThanOrEqual,
+            Operators.LengthLessThanOrEqual,
+          ].includes(adhocFilter.operatorId as Operators)
+            ? // Hide the Custom SQL tab for element-level array operators: they
+              // have no portable SQL representation, and converting one would
+              // silently turn the filter into invalid raw SQL.
+              []
            : [
                {
                  key: ExpressionTypes.Sql,
@@ -35,6 +35,7 @@ import {
 } from 'src/explore/constants';
 import AdhocMetric from 'src/explore/components/controls/MetricControl/AdhocMetric';
 import { FeatureFlag, isFeatureEnabled } from '@superset-ui/core';
+import { GenericDataType } from '@apache-superset/core/common';
 import fetchMock from 'fetch-mock';

 import { TestDataset, Dataset } from '@superset-ui/chart-controls';
@@ -252,6 +253,78 @@ test('shows boolean only operators when subject is number', () => {
  ].map(operator => expect(isOperatorRelevant(operator, 'value')).toBe(true));
 });

+test('shows array operators (tier 1 + tier 2) when subject is multi-value', () => {
+  const props = setup({
+    adhocFilter: new AdhocFilter({
+      expressionType: ExpressionTypes.Simple,
+      subject: 'skills',
+      operatorId: undefined,
+      operator: undefined,
+      comparator: undefined,
+      clause: undefined,
+    }),
+    datasource: {
+      columns: [
+        {
+          id: 3,
+          column_name: 'skills',
+          type: 'Array(String)',
+          type_generic: GenericDataType.MultiValue,
+        },
+      ],
+    },
+  });
+  const { isOperatorRelevant } = useSimpleTabFilterProps(
+    props as unknown as Props,
+  );
+  // Tier 1 (whole-array) + Tier 2 (element-level) are all relevant.
+  [
+    Operators.Equals,
+    Operators.NotEquals,
+    Operators.In,
+    Operators.NotIn,
+    Operators.IsNull,
+    Operators.IsNotNull,
+    Operators.ContainsAny,
+    Operators.ContainsAll,
+    Operators.IsEmpty,
+    Operators.IsNotEmpty,
+  ].forEach(operator =>
+    expect(isOperatorRelevant(operator, 'skills')).toBe(true),
+  );
+  // scalar-only operators are hidden for array columns
+  [Operators.GreaterThan, Operators.LessThan, Operators.Like].forEach(
+    operator => expect(isOperatorRelevant(operator, 'skills')).toBe(false),
+  );
+});
+
+test('hides element-level array operators for non multi-value columns', () => {
+  const props = setup({
+    adhocFilter: new AdhocFilter({
+      expressionType: ExpressionTypes.Simple,
+      subject: 'value',
+      operatorId: undefined,
+      operator: undefined,
+      comparator: undefined,
+      clause: undefined,
+    }),
+    datasource: {
+      columns: [{ id: 3, column_name: 'value', type: 'STRING' }],
+    },
+  });
+  const { isOperatorRelevant } = useSimpleTabFilterProps(
+    props as unknown as Props,
+  );
+  [
+    Operators.ContainsAny,
+    Operators.ContainsAll,
+    Operators.IsEmpty,
+    Operators.IsNotEmpty,
+  ].forEach(operator =>
+    expect(isOperatorRelevant(operator, 'value')).toBe(false),
+  );
+});
+
 test('will convert from individual comparator to array if the operator changes to multi', () => {
  const props = setup();
  const { onOperatorChange } = useSimpleTabFilterProps(
@@ -309,6 +382,49 @@ test('will convert from array to individual comparators if the operator changes
  );
 });

+test('resets the comparator when switching between array value families', () => {
+  // Equal to (whole-array literal) -> Contains all (individual elements):
+  // the value spaces are incompatible, so the stale value must be cleared.
+  const wholeArrayFilter = new AdhocFilter({
+    expressionType: ExpressionTypes.Simple,
+    subject: 'scores',
+    operatorId: Operators.Equals,
+    operator: OPERATOR_ENUM_TO_OPERATOR_TYPE[Operators.Equals].operation,
+    comparator: '[5,6,7]',
+    clause: Clauses.Where,
+  });
+  const props = setup({ adhocFilter: wholeArrayFilter });
+  const { onOperatorChange } = useSimpleTabFilterProps(
+    props as unknown as Props,
+  );
+  onOperatorChange(Operators.ContainsAll);
+  const lastCall =
+    props.onChange.mock.calls[props.onChange.mock.calls.length - 1][0];
+  expect(lastCall.operatorId).toEqual(Operators.ContainsAll);
+  expect(lastCall.comparator).toBeUndefined();
+});
+
+test('keeps the value when switching within the element family', () => {
+  // Contains any <-> Contains all both take individual elements, so the
+  // selected elements should carry over.
+  const elementFilter = new AdhocFilter({
+    expressionType: ExpressionTypes.Simple,
+    subject: 'scores',
+    operatorId: Operators.ContainsAny,
+    operator: OPERATOR_ENUM_TO_OPERATOR_TYPE[Operators.ContainsAny].operation,
+    comparator: ['5', '6'],
+    clause: Clauses.Where,
+  });
+  const props = setup({ adhocFilter: elementFilter });
+  const { onOperatorChange } = useSimpleTabFilterProps(
+    props as unknown as Props,
+  );
+  onOperatorChange(Operators.ContainsAll);
+  const lastCall =
+    props.onChange.mock.calls[props.onChange.mock.calls.length - 1][0];
+  expect(lastCall.comparator).toEqual(['5', '6']);
+});
+
 test('passes the new adhocFilter to onChange after onComparatorChange', () => {
  const props = setup();
  const { onComparatorChange } = useSimpleTabFilterProps(
@@ -399,6 +515,28 @@ test('will not display boolean operators when column type is string', () => {
  });
 });

+test.each(['STRING', 'DATE'])(
+  'will not display boolean operators when an expression column declares type %s',
+  type => {
+    const props = setup({
+      datasource: {
+        type: 'table' as const,
+        datasource_name: 'table1',
+        schema: 'schema',
+        columns: [{ column_name: 'value', type, expression: '"value"' }],
+      },
+      adhocFilter: simpleAdhocFilter,
+    });
+    const { isOperatorRelevant } = useSimpleTabFilterProps(
+      props as unknown as Props,
+    );
+    const booleanOnlyOperators = [Operators.IsTrue, Operators.IsFalse];
+    booleanOnlyOperators.forEach(operator => {
+      expect(isOperatorRelevant(operator, 'value')).toBe(false);
+    });
+  },
+);
+
 test('will display boolean operators when column is an expression', () => {
  const props = setup({
    datasource: {
@@ -32,6 +32,7 @@ import {
  isDefined,
  SupersetClient,
 } from '@superset-ui/core';
+import { GenericDataType } from '@apache-superset/core/common';
 import { styled, useTheme, css } from '@apache-superset/core/theme';
 import {
  Operators,
@@ -118,6 +119,8 @@ export const useSimpleTabFilterProps = (props: Props) => {
    const isColumnNumber =
      !!column && (column.type === 'INT' || column.type === 'INTEGER');
    const isColumnFunction = !!column && !!column.expression;
+    const isColumnMultiValue =
+      !!column && column.type_generic === GenericDataType.MultiValue;

    if (operator && operator === Operators.LatestPartition) {
      const { partitionColumn } = props;
@@ -127,8 +130,41 @@ export const useSimpleTabFilterProps = (props: Props) => {
      // hide the TEMPORAL_RANGE operator
      return false;
    }
+    // Element-level array operators only apply to multi-value columns.
+    const arrayElementOperators = [
+      Operators.ContainsAny,
+      Operators.ContainsAll,
+      Operators.IsEmpty,
+      Operators.IsNotEmpty,
+      Operators.LengthEquals,
+      Operators.LengthGreaterThan,
+      Operators.LengthLessThan,
+      Operators.LengthGreaterThanOrEqual,
+      Operators.LengthLessThanOrEqual,
+    ];
+    if (arrayElementOperators.includes(operator)) {
+      return isColumnMultiValue;
+    }
+    if (isColumnMultiValue) {
+      // Array columns support whole-array operators (=, !=, In, Not in, null
+      // checks) plus the element-level operators above. Scalar-only operators
+      // (Like, <, >, <=, >=) are hidden because they aren't valid on an array.
+      return [
+        Operators.Equals,
+        Operators.NotEquals,
+        Operators.In,
+        Operators.NotIn,
+        Operators.IsNull,
+        Operators.IsNotNull,
+        ...arrayElementOperators,
+      ].includes(operator);
+    }
    if (operator === Operators.IsTrue || operator === Operators.IsFalse) {
-      return isColumnBoolean || isColumnNumber || isColumnFunction;
+      // An expression column may evaluate to a boolean, but that is only a
+      // safe assumption while its type is unknown; a declared type wins.
+      return (
+        isColumnBoolean || isColumnNumber || (isColumnFunction && !column?.type)
+      );
    }
    if (isColumnBoolean) {
      return operator === Operators.IsNull || operator === Operators.IsNotNull;
@@ -167,9 +203,19 @@ export const useSimpleTabFilterProps = (props: Props) => {
          ].operation
        : null;
    if (!isDefined(operator)) {
-      // if operator is `null`, use the `IN` and reset the comparator.
-      operator = Operators.In;
-      operatorId = Operators.In;
+      // The previous operator is not relevant for the new subject; pick a
+      // sensible default and reset the comparator. Multi-value (array) columns
+      // default to "Contains any" (element membership) rather than the
+      // scalar-only IN.
+      const newColumn = props.datasource.columns?.find(
+        col => col.column_name === subject,
+      );
+      const defaultOperator =
+        newColumn?.type_generic === GenericDataType.MultiValue
+          ? Operators.ContainsAny
+          : Operators.In;
+      operator = defaultOperator;
+      operatorId = defaultOperator;
      comparator = undefined;
    }

@@ -193,10 +239,38 @@ export const useSimpleTabFilterProps = (props: Props) => {
  };
  const onOperatorChange = (operatorId: Operators) => {
    const currentComparator = props.adhocFilter.comparator;
+    // The value space differs between operator families: element-level array
+    // ops (Contains any/all) take individual elements, whole-array/scalar ops
+    // (=, In, …) take whole arrays or scalars, Length ops take a count, and the
+    // unary ops take nothing. A value from one family is meaningless in another,
+    // so reset the value when the family changes (e.g. Equal to -> Contains all).
+    const comparatorKind = (op?: Operators): string => {
+      if (!op) return 'none';
+      if (op === Operators.ContainsAny || op === Operators.ContainsAll) {
+        return 'element';
+      }
+      if (
+        op === Operators.LengthEquals ||
+        op === Operators.LengthGreaterThan ||
+        op === Operators.LengthLessThan ||
+        op === Operators.LengthGreaterThanOrEqual ||
+        op === Operators.LengthLessThanOrEqual
+      ) {
+        return 'length';
+      }
+      if (DISABLE_INPUT_OPERATORS.includes(op)) return 'none';
+      return 'value';
+    };
+    const valueFamilyChanged =
+      comparatorKind(props.adhocFilter.operatorId as Operators | undefined) !==
+      comparatorKind(operatorId);
+
    let newComparator;
-    // convert between list of comparators and individual comparators
-    // (e.g. `in ('North America', 'Africa')` to `== 'North America'`)
-    if (MULTI_OPERATORS.has(operatorId)) {
+    if (valueFamilyChanged) {
+      newComparator = undefined;
+    } else if (MULTI_OPERATORS.has(operatorId)) {
+      // convert between list of comparators and individual comparators
+      // (e.g. `in ('North America', 'Africa')` to `== 'North America'`)
      newComparator = Array.isArray(currentComparator)
        ? currentComparator
        : [currentComparator].filter(element => element != null);
@@ -433,19 +507,42 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
        if (loadingComparatorSuggestions) {
          controller.abort();
        }
+        // Element-level array operators (Contains any / Contains all) search
+        // inside the array, so suggest individual elements; whole-array
+        // operators (=, In, …) keep the default distinct-array suggestions.
+        const { operatorId } = props.adhocFilter;
+        const arrayElements =
+          operatorId === Operators.ContainsAny ||
+          operatorId === Operators.ContainsAll;
        setLoadingComparatorSuggestions(true);
        SupersetClient.get({
          signal,
-          endpoint: `/api/v1/datasource/${datasource.type}/${datasource.id}/column/${col}/values/`,
+          endpoint: `/api/v1/datasource/${datasource.type}/${datasource.id}/column/${col}/values/${
+            arrayElements ? '?array_elements=true' : ''
+          }`,
        })
          .then(({ json }) => {
            setSuggestions(
-              json.result.map(
-                (suggestion: null | number | boolean | string) => ({
-                  value: suggestion,
-                  label: optionLabel(suggestion),
-                }),
-              ),
+              json.result.map((suggestion: unknown) => {
+                // Complex column values arrive as JS arrays or objects: whole
+                // arrays for MULTI_VALUE columns (e.g. [5, 6, 7]) and Map/Tuple
+                // objects for nested-container columns (e.g. {"a": ["x","y"]}).
+                // A raw array/object is neither a valid single-select value
+                // (antd collapses an array to its first element) nor renderable
+                // as a React child (an object throws). Render it as its literal
+                // string, which is also exactly what the backend's
+                // parse_array_literal expects for the whole-array operators.
+                if (suggestion !== null && typeof suggestion === 'object') {
+                  const literal = JSON.stringify(suggestion);
+                  return { value: literal, label: literal };
+                }
+                return {
+                  value: suggestion as null | number | boolean | string,
+                  label: optionLabel(
+                    suggestion as null | number | boolean | string,
+                  ),
+                };
+              }),
            );
            setLoadingComparatorSuggestions(false);
          })
@@ -464,6 +561,7 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
  }, [
    props.adhocFilter.subject,
    props.adhocFilter.clause,
+    props.adhocFilter.operatorId,
    props.datasource,
    datePicker,
  ]);
@@ -44,6 +44,17 @@ export const OPERATORS_TO_SQL = {
  'IS NULL': 'IS NULL',
  'IS TRUE': 'IS TRUE',
  'IS FALSE': 'IS FALSE',
+  // Element-level array operators (shown as filter labels; not executable SQL —
+  // the Custom SQL tab is hidden for these).
+  CONTAINS_ANY: 'CONTAINS ANY',
+  CONTAINS_ALL: 'CONTAINS ALL',
+  IS_EMPTY: 'IS EMPTY',
+  IS_NOT_EMPTY: 'IS NOT EMPTY',
+  LENGTH_EQUALS: 'LENGTH =',
+  LENGTH_GREATER_THAN: 'LENGTH >',
+  LENGTH_LESS_THAN: 'LENGTH <',
+  LENGTH_GREATER_THAN_OR_EQUALS: 'LENGTH >=',
+  LENGTH_LESS_THAN_OR_EQUALS: 'LENGTH <=',
  'LATEST PARTITION': ({
    datasource,
  }: {
@@ -75,6 +75,7 @@ import ViewQueryModal from '../controls/ViewQueryModal';
 import EmbedCodeContent from '../EmbedCodeContent';
 import { useDashboardsMenuItems } from './DashboardsSubMenu';
 import { useExploreDataExport } from './useExploreDataExport';
+import { LineageModal } from 'src/features/lineage';

 export const SEARCH_THRESHOLD = 10;

@@ -111,6 +112,7 @@ const MENU_KEYS = {
  EDIT_REPORT: 'edit_report',
  DELETE_REPORT: 'delete_report',
  VIEW_QUERY: 'view_query',
+  VIEW_LINEAGE: 'view_lineage',
  RUN_IN_SQL_LAB: 'run_in_sql_lab',
  VERSION_HISTORY: 'version_history',
 };
@@ -1079,6 +1081,23 @@ export const useExploreAdditionalActionsMenu = (
      onClick: () => setIsDropdownVisible(false),
    });

+    // View lineage
+    if (slice?.slice_id) {
+      menuItems.push({
+        key: MENU_KEYS.VIEW_LINEAGE,
+        label: (
+          <LineageModal
+            entityType="chart"
+            entityId={slice.slice_id}
+            triggerNode={
+              <div data-test="view-lineage-menu-item">{t('View lineage')}</div>
+            }
+          />
+        ),
+        onClick: () => setIsDropdownVisible(false),
+      });
+    }
+
    // Run in SQL Lab
    if (datasource) {
      menuItems.push({
@@ -45,6 +45,17 @@ export enum Operators {
  IsTrue = 'IS_TRUE',
  IsFalse = 'IS_FALSE',
  TemporalRange = 'TEMPORAL_RANGE',
+  // Element-level operators for multi-value (array) columns
+  ContainsAny = 'CONTAINS_ANY',
+  ContainsAll = 'CONTAINS_ALL',
+  IsEmpty = 'IS_EMPTY',
+  IsNotEmpty = 'IS_NOT_EMPTY',
+  // Length (element-count) comparison operators for array columns
+  LengthEquals = 'LENGTH_EQUALS',
+  LengthGreaterThan = 'LENGTH_GREATER_THAN',
+  LengthLessThan = 'LENGTH_LESS_THAN',
+  LengthGreaterThanOrEqual = 'LENGTH_GREATER_THAN_OR_EQUALS',
+  LengthLessThanOrEqual = 'LENGTH_LESS_THAN_OR_EQUALS',
 }

 export interface OperatorType {
@@ -89,6 +100,39 @@ export const OPERATOR_ENUM_TO_OPERATOR_TYPE: {
    display: t('TEMPORAL_RANGE'),
    operation: 'TEMPORAL_RANGE',
  },
+  [Operators.ContainsAny]: {
+    display: t('Contains any'),
+    operation: 'CONTAINS_ANY',
+  },
+  [Operators.ContainsAll]: {
+    display: t('Contains all'),
+    operation: 'CONTAINS_ALL',
+  },
+  [Operators.IsEmpty]: { display: t('Is empty'), operation: 'IS_EMPTY' },
+  [Operators.IsNotEmpty]: {
+    display: t('Is not empty'),
+    operation: 'IS_NOT_EMPTY',
+  },
+  [Operators.LengthEquals]: {
+    display: t('Length equals (=)'),
+    operation: 'LENGTH_EQUALS',
+  },
+  [Operators.LengthGreaterThan]: {
+    display: t('Length greater than (>)'),
+    operation: 'LENGTH_GREATER_THAN',
+  },
+  [Operators.LengthLessThan]: {
+    display: t('Length less than (<)'),
+    operation: 'LENGTH_LESS_THAN',
+  },
+  [Operators.LengthGreaterThanOrEqual]: {
+    display: t('Length greater or equal (>=)'),
+    operation: 'LENGTH_GREATER_THAN_OR_EQUALS',
+  },
+  [Operators.LengthLessThanOrEqual]: {
+    display: t('Length less or equal (<=)'),
+    operation: 'LENGTH_LESS_THAN_OR_EQUALS',
+  },
 };

 export const OPERATORS_OPTIONS = Object.values(Operators) as Operators[];
@@ -105,7 +149,12 @@ export const HAVING_OPERATORS = [
  Operators.GreaterThan,
  Operators.GreaterThanOrEqual,
 ];
-export const MULTI_OPERATORS = new Set([Operators.In, Operators.NotIn]);
+export const MULTI_OPERATORS = new Set([
+  Operators.In,
+  Operators.NotIn,
+  Operators.ContainsAny,
+  Operators.ContainsAll,
+]);
 // CUSTOM_OPERATORS will show operator in simple mode,
 // but will generate customized sqlExpression
 export const CUSTOM_OPERATORS = new Set([
@@ -120,6 +169,8 @@ export const DISABLE_INPUT_OPERATORS = [
  Operators.LatestPartition,
  Operators.IsTrue,
  Operators.IsFalse,
+  Operators.IsEmpty,
+  Operators.IsNotEmpty,
 ];

 export const sqlaAutoGeneratedMetricNameRegex =
@@ -82,3 +82,14 @@ test('Should handle boolean true comparator as a string value', () => {
    "subject operator 'TRUE'",
  );
 });
+
+test('Should render array-literal comparators as-is (not quoted)', () => {
+  // Whole-array = filter: the pasted array literal is shown unquoted.
+  expect(getSimpleSQLExpression('ingredients', '=', "['1 large egg']")).toBe(
+    "ingredients = ['1 large egg']",
+  );
+  // IN with multiple array literals.
+  expect(
+    getSimpleSQLExpression('ingredients', Operators.In, ["['a']", "['b']"]),
+  ).toBe(`ingredients ${Operators.In} (['a'], ['b'])`);
+});
@@ -461,10 +461,15 @@ export const getSimpleSQLExpression = (
    if (comparatorArray.length > 0 && showComparator) {
      const formattedComparators = comparatorArray
        .map(val => optionLabel(val))
-        .map(
-          val =>
-            `${quote}${isString ? String(val).replace(/'/g, "''") : val}${quote}`,
-        );
+        .map(val => {
+          // Array-literal values (e.g. ['a', 'b']) are shown as-is rather than
+          // quoted/escaped as a string, so array-column filters read naturally.
+          const asString = String(val);
+          if (asString.startsWith('[') && asString.endsWith(']')) {
+            return asString;
+          }
+          return `${quote}${isString ? asString.replace(/'/g, "''") : val}${quote}`;
+        });
      expression += ` ${prefix}${formattedComparators.join(', ')}${suffix}`;
    }
  }
@@ -38,6 +38,7 @@ import type { ListViewFetchDataConfig as FetchDataConfig } from 'src/components'
 import { TableTab } from 'src/views/CRUD/types';
 import { isUserEditorOrAdmin } from 'src/dashboard/util/permissionUtils';
 import type { UserWithPermissionsAndRoles } from 'src/types/bootstrapTypes';
+import { LineageModal } from 'src/features/lineage';

 const menuItemButtonCss = css`
  appearance: none;
@@ -93,6 +94,7 @@ export default function ChartCard({
  const canDelete = hasPerm('can_write');
  const canExport = hasPerm('can_export');
  const allowEdit = isUserEditorOrAdmin(user, chart.editors);
+  const canRead = hasPerm('can_read');
  const menuItems: MenuItem[] = [];

  if (canEdit) {
@@ -128,6 +130,29 @@ export default function ChartCard({
    });
  }

+  if (canRead) {
+    menuItems.push({
+      key: 'lineage',
+      label: (
+        <LineageModal
+          entityType="chart"
+          entityId={chart.id}
+          triggerNode={
+            <div>
+              <Icons.ShareAltOutlined
+                iconSize="l"
+                css={css`
+                  vertical-align: text-top;
+                `}
+              />{' '}
+              {t('View Lineage')}
+            </div>
+          }
+        />
+      ),
+    });
+  }
+
  if (canExport) {
    menuItems.push({
      key: 'export',
@@ -36,6 +36,7 @@ import { KebabMenuButton } from 'src/components';
 import { isUserEditorOrAdmin } from 'src/dashboard/util/permissionUtils';
 import type { UserWithPermissionsAndRoles } from 'src/types/bootstrapTypes';
 import { useIsMobile } from 'src/hooks/useIsMobile';
+import { LineageModal } from 'src/features/lineage';

 const menuItemButtonCss = css`
  appearance: none;
@@ -84,6 +85,7 @@ function DashboardCard({
  const canDelete = hasPerm('can_write');
  const canExport = hasPerm('can_export');
  const allowEdit = isUserEditorOrAdmin(user, dashboard.editors);
+  const canRead = hasPerm('can_read');
  const digest = dashboard.changed_on_utc || dashboard.changed_on;
  const thumbnailUrl =
    isFeatureEnabled(FeatureFlag.Thumbnails) && dashboard.id && digest
@@ -92,6 +94,23 @@ function DashboardCard({

  const menuItems: MenuItem[] = [];

+  if (canRead) {
+    menuItems.push({
+      key: 'lineage',
+      label: (
+        <LineageModal
+          entityType="dashboard"
+          entityId={dashboard.id}
+          triggerNode={
+            <div data-test="dashboard-card-option-lineage-button">
+              <Icons.ShareAltOutlined iconSize="l" /> {t('View Lineage')}
+            </div>
+          }
+        />
+      ),
+    });
+  }
+
  if (canEdit && openDashboardEditModal) {
    menuItems.push({
      key: 'edit',
@@ -16,11 +16,14 @@
 * specific language governing permissions and limitations
 * under the License.
 */
+import { useState } from 'react';
 import { t } from '@apache-superset/core/translation';
 import { styled } from '@apache-superset/core/theme';
 import useGetDatasetRelatedCounts from 'src/features/datasets/hooks/useGetDatasetRelatedCounts';
 import { Badge } from '@superset-ui/core/components';
 import Tabs from '@superset-ui/core/components/Tabs';
+import { useDatasetLineage } from 'src/hooks/apiResources';
+import { LineageView } from 'src/features/lineage';

 const StyledTabs = styled(Tabs)`
  ${({ theme }) => `
@@ -51,16 +54,25 @@ const TRANSLATIONS = {
  USAGE_TEXT: t('Usage'),
  COLUMNS_TEXT: t('Columns'),
  METRICS_TEXT: t('Metrics'),
+  LINEAGE_TEXT: t('Lineage'),
 };

 const TABS_KEYS = {
  COLUMNS: 'COLUMNS',
  METRICS: 'METRICS',
  USAGE: 'USAGE',
+  LINEAGE: 'LINEAGE',
 };

 const EditPage = ({ id }: EditPageProps) => {
  const { usageCount } = useGetDatasetRelatedCounts(id);
+  const [activeKey, setActiveKey] = useState(TABS_KEYS.COLUMNS);
+  // Only fetch lineage once the user opens the Lineage tab to avoid
+  // unnecessary requests/backend load on page load.
+  const lineageResource = useDatasetLineage(
+    id,
+    activeKey !== TABS_KEYS.LINEAGE,
+  );

  const usageTab = (
    <TabStyles>
@@ -85,9 +97,23 @@ const EditPage = ({ id }: EditPageProps) => {
      label: usageTab,
      children: null,
    },
+    {
+      key: TABS_KEYS.LINEAGE,
+      label: TRANSLATIONS.LINEAGE_TEXT,
+      children: (
+        <LineageView lineageResource={lineageResource} entityType="dataset" />
+      ),
+    },
  ];

-  return <StyledTabs moreIcon={null} items={items} />;
+  return (
+    <StyledTabs
+      moreIcon={null}
+      items={items}
+      activeKey={activeKey}
+      onChange={setActiveKey}
+    />
+  );
 };

 export default EditPage;
@@ -0,0 +1,90 @@
+/**
+ * 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.
+ */
+import { FC, ReactNode, useState } from 'react';
+import { t } from '@apache-superset/core/translation';
+import { ModalTrigger } from '@superset-ui/core/components';
+import {
+  useChartLineage,
+  useDashboardLineage,
+  useDatasetLineage,
+} from 'src/hooks/apiResources';
+import LineageView from './LineageView';
+
+export interface LineageModalProps {
+  entityType: 'dataset' | 'chart' | 'dashboard';
+  entityId: string | number;
+  triggerNode: ReactNode;
+}
+
+const LineageModal: FC<LineageModalProps> = ({
+  entityType,
+  entityId,
+  triggerNode,
+}) => {
+  // Defer the lineage fetch until the modal is actually opened so that simply
+  // rendering the trigger (e.g. inside an actions dropdown) does not hit the
+  // lineage endpoint.
+  const [opened, setOpened] = useState(false);
+
+  const datasetLineage = useDatasetLineage(
+    entityType === 'dataset' ? entityId : '',
+    !opened,
+  );
+  const chartLineage = useChartLineage(
+    entityType === 'chart' ? entityId : '',
+    !opened,
+  );
+  const dashboardLineage = useDashboardLineage(
+    entityType === 'dashboard' ? entityId : '',
+    !opened,
+  );
+
+  const lineageResource =
+    entityType === 'dataset'
+      ? datasetLineage
+      : entityType === 'chart'
+        ? chartLineage
+        : dashboardLineage;
+
+  const title =
+    entityType === 'dataset'
+      ? t('Dataset Lineage')
+      : entityType === 'chart'
+        ? t('Chart Lineage')
+        : t('Dashboard Lineage');
+
+  return (
+    <ModalTrigger
+      triggerNode={triggerNode}
+      beforeOpen={() => setOpened(true)}
+      modalTitle={title}
+      modalBody={
+        <LineageView
+          lineageResource={lineageResource}
+          entityType={entityType}
+        />
+      }
+      width="850px"
+      responsive
+      destroyOnHidden
+    />
+  );
+};
+
+export default LineageModal;
@@ -0,0 +1,743 @@
+/**
+ * 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.
+ */
+import { FC, useMemo, useState, useCallback } from 'react';
+import { useResizeDetector } from 'react-resize-detector';
+import { t } from '@apache-superset/core/translation';
+import { styled, useTheme } from '@apache-superset/core/theme';
+import { Empty, Loading } from '@superset-ui/core/components';
+import { Button } from '@superset-ui/core/components';
+import { ResourceStatus } from 'src/hooks/apiResources/apiResources';
+import type { Resource } from 'src/hooks/apiResources/apiResources';
+import type {
+  DatasetLineage,
+  ChartLineage,
+  DashboardLineage,
+  ChartEntity,
+  DashboardEntity,
+  DatasetEntity,
+  DatabaseEntity,
+} from 'src/hooks/apiResources/lineage';
+import Echart from '../../../plugins/plugin-chart-echarts/src/components/Echart';
+import type { EChartsCoreOption } from 'echarts/core';
+
+const LineageContainer = styled.div`
+  display: flex;
+  flex-direction: column;
+  width: 100%;
+  height: 100%;
+`;
+
+const Legend = styled.div`
+  ${({ theme }) => `
+    display: flex;
+    justify-content: center;
+    align-items: center;
+    gap: ${theme.sizeUnit * 4}px;
+    padding: ${theme.sizeUnit * 3}px;
+    background-color: ${theme.colorBgLayout};
+    border-bottom: 1px solid ${theme.colorBorder};
+  `}
+`;
+
+const LegendItem = styled.div<{ color: string }>`
+  ${({ theme, color }) => `
+    display: flex;
+    align-items: center;
+    gap: ${theme.sizeUnit * 2}px;
+    font-size: ${theme.fontSizeSM}px;
+    color: ${theme.colorText};
+
+    &::before {
+      content: '';
+      width: 12px;
+      height: 12px;
+      border-radius: 2px;
+      background-color: ${color};
+    }
+  `}
+`;
+
+const DetailsPanel = styled.div`
+  ${({ theme }) => `
+    padding: ${theme.sizeUnit * 4}px;
+    background-color: ${theme.colorBgLayout};
+    border-top: 1px solid ${theme.colorBorder};
+    min-height: 120px;
+  `}
+`;
+
+const DetailsPanelHeader = styled.div`
+  ${({ theme }) => `
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+    margin-bottom: ${theme.sizeUnit * 3}px;
+  `}
+`;
+
+const DetailsPanelActions = styled.div`
+  ${({ theme }) => `
+    display: flex;
+    gap: ${theme.sizeUnit * 2}px;
+  `}
+`;
+
+const DetailsPanelTitle = styled.h4`
+  ${({ theme }) => `
+    margin: 0;
+    font-size: ${theme.fontSizeLG}px;
+    font-weight: ${theme.fontWeightStrong};
+    color: ${theme.colorText};
+  `}
+`;
+
+const DetailsPanelContent = styled.div`
+  ${({ theme }) => `
+    display: flex;
+    flex-direction: column;
+    gap: ${theme.sizeUnit * 2}px;
+  `}
+`;
+
+const DetailRow = styled.div`
+  ${({ theme }) => `
+    display: flex;
+    gap: ${theme.sizeUnit * 2}px;
+    font-size: ${theme.fontSizeSM}px;
+    color: ${theme.colorText};
+  `}
+`;
+
+const DetailLabel = styled.span`
+  ${({ theme }) => `
+    font-weight: ${theme.fontWeightStrong};
+    min-width: 100px;
+  `}
+`;
+
+const DetailValue = styled.span`
+  ${({ theme }) => `
+    color: ${theme.colorTextSecondary};
+  `}
+`;
+
+type NodeType = 'database' | 'dataset' | 'chart' | 'dashboard';
+
+type NodeDetails = {
+  name: string;
+  type: NodeType;
+  id?: number;
+  additionalInfo?: Record<string, string | number | null | undefined>;
+};
+
+// Build a stable, unique graph identity for a node so that entities sharing the
+// same display name (e.g. two charts with identical titles) never collapse into
+// a single Sankey node. The human-readable name is kept separately as the label.
+const nodeKey = (type: NodeType, id?: number, name?: string): string =>
+  id != null ? `${type}:${id}` : `${type}:${name ?? ''}`;
+
+type LineageViewProps = {
+  lineageResource:
+    | Resource<DatasetLineage>
+    | Resource<ChartLineage>
+    | Resource<DashboardLineage>;
+  entityType: 'dataset' | 'chart' | 'dashboard';
+};
+
+const LineageView: FC<LineageViewProps> = ({ lineageResource, entityType }) => {
+  const theme = useTheme();
+  const [selectedNode, setSelectedNode] = useState<NodeDetails | null>(null);
+  const { width: chartWidth = 800, ref: chartContainerRef } =
+    useResizeDetector();
+
+  // Create a mapping of node names to their details
+  const nodeDetailsMap = useMemo(() => {
+    if (
+      lineageResource.status !== ResourceStatus.Complete ||
+      !lineageResource.result
+    ) {
+      return new Map<string, NodeDetails>();
+    }
+
+    const data = lineageResource.result;
+    const map = new Map<string, NodeDetails>();
+
+    if (entityType === 'dataset' && 'dataset' in data) {
+      const { dataset, upstream, downstream } = data as DatasetLineage;
+
+      // Add current dataset
+      map.set(nodeKey('dataset', dataset.id, dataset.name), {
+        name: dataset.name,
+        type: 'dataset',
+        id: dataset.id,
+        additionalInfo: {
+          schema: dataset.schema,
+          table_name: dataset.table_name,
+          database_name: dataset.database_name,
+        },
+      });
+
+      // Add upstream database
+      if (upstream?.database) {
+        map.set(
+          nodeKey(
+            'database',
+            upstream.database.id,
+            upstream.database.database_name,
+          ),
+          {
+            name: upstream.database.database_name,
+            type: 'database',
+            id: upstream.database.id,
+          },
+        );
+      }
+
+      // Add downstream charts
+      if (downstream?.charts?.result) {
+        downstream.charts.result.forEach((chart: ChartEntity) => {
+          map.set(nodeKey('chart', chart.id, chart.slice_name), {
+            name: chart.slice_name,
+            type: 'chart',
+            id: chart.id,
+            additionalInfo: {
+              viz_type: chart.viz_type,
+            },
+          });
+        });
+      }
+
+      // Add downstream dashboards
+      if (downstream?.dashboards?.result) {
+        downstream.dashboards.result.forEach((dashboard: DashboardEntity) => {
+          map.set(nodeKey('dashboard', dashboard.id, dashboard.title), {
+            name: dashboard.title,
+            type: 'dashboard',
+            id: dashboard.id,
+            additionalInfo: {
+              slug: dashboard.slug,
+            },
+          });
+        });
+      }
+    } else if (entityType === 'chart' && 'chart' in data) {
+      const { chart, upstream, downstream } = data as ChartLineage;
+
+      // Add current chart
+      map.set(nodeKey('chart', chart.id, chart.slice_name), {
+        name: chart.slice_name,
+        type: 'chart',
+        id: chart.id,
+        additionalInfo: {
+          viz_type: chart.viz_type,
+        },
+      });
+
+      // Add upstream dataset
+      if (upstream?.dataset) {
+        map.set(
+          nodeKey('dataset', upstream.dataset.id, upstream.dataset.name),
+          {
+            name: upstream.dataset.name,
+            type: 'dataset',
+            id: upstream.dataset.id,
+            additionalInfo: {
+              schema: upstream.dataset.schema,
+              table_name: upstream.dataset.table_name,
+            },
+          },
+        );
+      }
+
+      // Add upstream database
+      if (upstream?.database) {
+        map.set(
+          nodeKey(
+            'database',
+            upstream.database.id,
+            upstream.database.database_name,
+          ),
+          {
+            name: upstream.database.database_name,
+            type: 'database',
+            id: upstream.database.id,
+          },
+        );
+      }
+
+      // Add downstream dashboards
+      if (downstream?.dashboards?.result) {
+        downstream.dashboards.result.forEach((dashboard: DashboardEntity) => {
+          map.set(nodeKey('dashboard', dashboard.id, dashboard.title), {
+            name: dashboard.title,
+            type: 'dashboard',
+            id: dashboard.id,
+            additionalInfo: {
+              slug: dashboard.slug,
+            },
+          });
+        });
+      }
+    } else if (entityType === 'dashboard' && 'dashboard' in data) {
+      const { dashboard, upstream } = data as DashboardLineage;
+
+      // Add current dashboard
+      map.set(nodeKey('dashboard', dashboard.id, dashboard.title), {
+        name: dashboard.title,
+        type: 'dashboard',
+        id: dashboard.id,
+        additionalInfo: {
+          slug: dashboard.slug,
+        },
+      });
+
+      // Add upstream charts
+      if (upstream?.charts?.result) {
+        upstream.charts.result.forEach((chart: ChartEntity) => {
+          map.set(nodeKey('chart', chart.id, chart.slice_name), {
+            name: chart.slice_name,
+            type: 'chart',
+            id: chart.id,
+            additionalInfo: {
+              viz_type: chart.viz_type,
+            },
+          });
+        });
+      }
+
+      // Add upstream datasets
+      if (upstream?.datasets?.result) {
+        upstream.datasets.result.forEach((dataset: DatasetEntity) => {
+          map.set(nodeKey('dataset', dataset.id, dataset.name), {
+            name: dataset.name,
+            type: 'dataset',
+            id: dataset.id,
+            additionalInfo: {
+              schema: dataset.schema,
+              table_name: dataset.table_name,
+            },
+          });
+        });
+      }
+
+      // Add upstream databases
+      if (upstream?.databases?.result) {
+        upstream.databases.result.forEach((database: DatabaseEntity) => {
+          map.set(nodeKey('database', database.id, database.database_name), {
+            name: database.database_name,
+            type: 'database',
+            id: database.id,
+          });
+        });
+      }
+    }
+
+    return map;
+  }, [lineageResource, entityType]);
+
+  // Handle node click
+  const handleNodeClick = useCallback(
+    (params: {
+      dataType?: string;
+      name?: string;
+      event?: { stop: () => void };
+    }) => {
+      if (params.dataType === 'node' && params.name) {
+        const nodeDetails = nodeDetailsMap.get(params.name);
+        if (nodeDetails) {
+          setSelectedNode(nodeDetails);
+        }
+      }
+      // Always stop event propagation to prevent tooltip issues
+      if (params.event) {
+        params.event.stop();
+      }
+    },
+    [nodeDetailsMap],
+  );
+
+  const echartOptions: EChartsCoreOption | null = useMemo(() => {
+    if (
+      lineageResource.status !== ResourceStatus.Complete ||
+      !lineageResource.result
+    ) {
+      return null;
+    }
+
+    const data = lineageResource.result;
+    const nodes: {
+      name: string;
+      label?: { position?: string; formatter?: string };
+      itemStyle?: { color: string };
+    }[] = [];
+    const links: { source: string; target: string; value: number }[] = [];
+    const nodeSet = new Set<string>();
+
+    // Helper to add a node. `key` is the stable unique identity used for graph
+    // links and detail lookups; `label` is the human-readable text shown.
+    const addNode = (
+      key: string,
+      label: string,
+      color: string,
+      labelPosition: 'left' | 'right' | 'inside',
+    ) => {
+      if (!nodeSet.has(key)) {
+        nodeSet.add(key);
+        nodes.push({
+          name: key,
+          itemStyle: { color },
+          label: {
+            position: labelPosition,
+            formatter: label,
+          },
+        });
+      }
+    };
+
+    // Helper to add a link between two node keys
+    const addLink = (source: string, target: string) => {
+      links.push({ source, target, value: 1 });
+    };
+
+    // Build nodes and links based on entity type
+    if (entityType === 'dataset' && 'dataset' in data) {
+      const { dataset, upstream, downstream } = data as DatasetLineage;
+
+      const datasetKey = nodeKey('dataset', dataset.id, dataset.name);
+      // Add current dataset node (center) - label inside
+      addNode(datasetKey, dataset.name, theme.colorPrimary, 'inside');
+
+      // Add upstream database - label on left
+      if (upstream?.database) {
+        const dbKey = nodeKey(
+          'database',
+          upstream.database.id,
+          upstream.database.database_name,
+        );
+        addNode(
+          dbKey,
+          upstream.database.database_name,
+          theme.colorInfo,
+          'left',
+        );
+        addLink(dbKey, datasetKey);
+      }
+
+      // Add downstream charts - label on right
+      const chartKeys = new Map<number, string>();
+      if (downstream?.charts?.result) {
+        downstream.charts.result.forEach((chart: ChartEntity) => {
+          const chartKey = nodeKey('chart', chart.id, chart.slice_name);
+          chartKeys.set(chart.id, chartKey);
+          addNode(chartKey, chart.slice_name, theme.colorSuccess, 'right');
+          addLink(datasetKey, chartKey);
+        });
+      }
+
+      // Add downstream dashboards - label on right
+      if (downstream?.dashboards?.result) {
+        downstream.dashboards.result.forEach((dashboard: DashboardEntity) => {
+          const dashKey = nodeKey('dashboard', dashboard.id, dashboard.title);
+          addNode(dashKey, dashboard.title, theme.colorWarning, 'right');
+
+          // Link from charts to dashboards using chart_ids
+          let linked = false;
+          if (dashboard.chart_ids && dashboard.chart_ids.length > 0) {
+            dashboard.chart_ids.forEach(chartId => {
+              const chartKey = chartKeys.get(chartId);
+              if (chartKey) {
+                addLink(chartKey, dashKey);
+                linked = true;
+              }
+            });
+          }
+          // Fall back to a direct dataset -> dashboard edge so the dashboard
+          // still renders when none of its charts resolve to visible nodes
+          // (e.g. after permission filtering); Sankey omits orphaned nodes.
+          if (!linked) {
+            addLink(datasetKey, dashKey);
+          }
+        });
+      }
+    } else if (entityType === 'chart' && 'chart' in data) {
+      const { chart, upstream, downstream } = data as ChartLineage;
+
+      const chartKey = nodeKey('chart', chart.id, chart.slice_name);
+      // Add current chart node (center) - label inside
+      addNode(chartKey, chart.slice_name, theme.colorPrimary, 'inside');
+
+      // Add upstream dataset - label on left
+      if (upstream?.dataset) {
+        const datasetKey = nodeKey(
+          'dataset',
+          upstream.dataset.id,
+          upstream.dataset.name,
+        );
+        addNode(datasetKey, upstream.dataset.name, theme.colorInfo, 'left');
+        addLink(datasetKey, chartKey);
+
+        // Add upstream database - label on left
+        if (upstream.database) {
+          const dbKey = nodeKey(
+            'database',
+            upstream.database.id,
+            upstream.database.database_name,
+          );
+          addNode(
+            dbKey,
+            upstream.database.database_name,
+            theme.colorWarning,
+            'left',
+          );
+          addLink(dbKey, datasetKey);
+        }
+      }
+
+      // Add downstream dashboards - label on right
+      if (downstream?.dashboards?.result) {
+        downstream.dashboards.result.forEach((dashboard: DashboardEntity) => {
+          const dashKey = nodeKey('dashboard', dashboard.id, dashboard.title);
+          addNode(dashKey, dashboard.title, theme.colorSuccess, 'right');
+          addLink(chartKey, dashKey);
+        });
+      }
+    } else if (entityType === 'dashboard' && 'dashboard' in data) {
+      const { dashboard, upstream } = data as DashboardLineage;
+
+      const dashKey = nodeKey('dashboard', dashboard.id, dashboard.title);
+      // Add current dashboard node (right) - label inside
+      addNode(dashKey, dashboard.title, theme.colorPrimary, 'inside');
+
+      // Add upstream charts - label on left
+      const chartKeys = new Map<number, string>();
+      if (upstream?.charts?.result) {
+        upstream.charts.result.forEach((chart: ChartEntity) => {
+          const chartKey = nodeKey('chart', chart.id, chart.slice_name);
+          chartKeys.set(chart.id, chartKey);
+          addNode(chartKey, chart.slice_name, theme.colorInfo, 'left');
+          addLink(chartKey, dashKey);
+        });
+      }
+
+      // Add upstream datasets - label on left
+      const datasetKeys = new Map<number, string>();
+      if (upstream?.datasets?.result) {
+        upstream.datasets.result.forEach(dataset => {
+          const datasetKey = nodeKey('dataset', dataset.id, dataset.name);
+          datasetKeys.set(dataset.id, datasetKey);
+          addNode(datasetKey, dataset.name, theme.colorSuccess, 'left');
+        });
+      }
+
+      // Link charts to their specific datasets using dataset_id from each chart
+      if (upstream?.charts?.result) {
+        upstream.charts.result.forEach((chart: ChartEntity) => {
+          if (chart.dataset_id) {
+            const datasetKey = datasetKeys.get(chart.dataset_id);
+            const chartKey = chartKeys.get(chart.id);
+            if (datasetKey && chartKey) {
+              addLink(datasetKey, chartKey);
+            }
+          }
+        });
+      }
+
+      // Add upstream databases and link to their specific datasets
+      if (upstream?.databases?.result) {
+        upstream.databases.result.forEach(database => {
+          const dbKey = nodeKey(
+            'database',
+            database.id,
+            database.database_name,
+          );
+          addNode(dbKey, database.database_name, theme.colorWarning, 'left');
+
+          // Link databases to datasets that belong to them using database_id
+          if (upstream.datasets?.result) {
+            upstream.datasets.result.forEach(dataset => {
+              if (dataset.database_id === database.id) {
+                const datasetKey = datasetKeys.get(dataset.id);
+                if (datasetKey) {
+                  addLink(dbKey, datasetKey);
+                }
+              }
+            });
+          }
+        });
+      }
+    }
+
+    return {
+      series: {
+        animation: false,
+        data: nodes,
+        lineStyle: {
+          color: 'source',
+        },
+        links,
+        type: 'sankey',
+      },
+      tooltip: {
+        show: false,
+      },
+    };
+  }, [lineageResource, entityType, theme]);
+
+  // Build legend data based on entity type
+  const legendItems: { label: string; color: string }[] = useMemo(() => {
+    if (entityType === 'dataset') {
+      return [
+        { label: t('Database (Upstream)'), color: theme.colorInfo },
+        { label: t('Dataset (Current)'), color: theme.colorPrimary },
+        { label: t('Chart (Downstream)'), color: theme.colorSuccess },
+        { label: t('Dashboard (Downstream)'), color: theme.colorWarning },
+      ];
+    } else if (entityType === 'chart') {
+      return [
+        { label: t('Database (Upstream)'), color: theme.colorWarning },
+        { label: t('Dataset (Upstream)'), color: theme.colorInfo },
+        { label: t('Chart (Current)'), color: theme.colorPrimary },
+        { label: t('Dashboard (Downstream)'), color: theme.colorSuccess },
+      ];
+    } else if (entityType === 'dashboard') {
+      return [
+        { label: t('Database (Upstream)'), color: theme.colorWarning },
+        { label: t('Dataset (Upstream)'), color: theme.colorSuccess },
+        { label: t('Chart (Upstream)'), color: theme.colorInfo },
+        { label: t('Dashboard (Current)'), color: theme.colorPrimary },
+      ];
+    }
+    return [];
+  }, [entityType, theme]);
+
+  if (lineageResource.status === ResourceStatus.Loading) {
+    return <Loading />;
+  }
+
+  if (
+    lineageResource.status === ResourceStatus.Error ||
+    !lineageResource.result
+  ) {
+    return <Empty description={t('Failed to load lineage data')} />;
+  }
+
+  if (!echartOptions) {
+    return <Empty description={t('No lineage data available')} />;
+  }
+
+  // Helper function to get the URL for an entity. Datasets have no standalone
+  // detail page, so only dashboards and charts expose an "Open" action.
+  const getEntityUrl = (nodeDetails: NodeDetails): string => {
+    switch (nodeDetails.type) {
+      case 'dashboard':
+        return `/superset/dashboard/${nodeDetails.id}/`;
+      case 'chart':
+        return `/explore/?slice_id=${nodeDetails.id}`;
+      default:
+        return '#';
+    }
+  };
+
+  return (
+    <LineageContainer>
+      <Legend>
+        {legendItems.map(item => (
+          <LegendItem key={item.label} color={item.color}>
+            {item.label}
+          </LegendItem>
+        ))}
+      </Legend>
+      <div ref={chartContainerRef} style={{ width: '100%' }}>
+        <Echart
+          refs={{}}
+          height={selectedNode ? 450 : 600}
+          width={chartWidth}
+          echartOptions={echartOptions}
+          vizType="sankey"
+          eventHandlers={{
+            click: handleNodeClick,
+          }}
+        />
+      </div>
+      {selectedNode && (
+        <DetailsPanel>
+          <DetailsPanelHeader>
+            <DetailsPanelTitle>
+              {t(
+                '%s Details',
+                selectedNode.type.charAt(0).toUpperCase() +
+                  selectedNode.type.slice(1),
+              )}
+            </DetailsPanelTitle>
+            <DetailsPanelActions>
+              {(selectedNode.type === 'dashboard' ||
+                selectedNode.type === 'chart') && (
+                <Button
+                  buttonStyle="primary"
+                  buttonSize="small"
+                  onClick={() => {
+                    window.location.href = getEntityUrl(selectedNode);
+                  }}
+                >
+                  {t('Open')}{' '}
+                  {selectedNode.type.charAt(0).toUpperCase() +
+                    selectedNode.type.slice(1)}
+                </Button>
+              )}
+              <Button
+                buttonStyle="tertiary"
+                buttonSize="small"
+                onClick={() => setSelectedNode(null)}
+              >
+                {t('Close')}
+              </Button>
+            </DetailsPanelActions>
+          </DetailsPanelHeader>
+          <DetailsPanelContent>
+            <DetailRow>
+              <DetailLabel>{t('Name')}:</DetailLabel>
+              <DetailValue>{selectedNode.name}</DetailValue>
+            </DetailRow>
+            {selectedNode.id && (
+              <DetailRow>
+                <DetailLabel>{t('ID')}:</DetailLabel>
+                <DetailValue>{selectedNode.id}</DetailValue>
+              </DetailRow>
+            )}
+            {selectedNode.additionalInfo &&
+              Object.entries(selectedNode.additionalInfo).map(
+                ([key, value]) => (
+                  <DetailRow key={key}>
+                    <DetailLabel>
+                      {key.charAt(0).toUpperCase() +
+                        key.slice(1).replace(/_/g, ' ')}
+                      :
+                    </DetailLabel>
+                    <DetailValue>{String(value)}</DetailValue>
+                  </DetailRow>
+                ),
+              )}
+          </DetailsPanelContent>
+        </DetailsPanel>
+      )}
+    </LineageContainer>
+  );
+};
+
+export default LineageView;
@@ -0,0 +1,21 @@
+/**
+ * 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.
+ */
+
+export { default as LineageView } from './LineageView';
+export { default as LineageModal } from './LineageModal';
@@ -31,7 +31,12 @@ import {
 import { Group, Role, UserObject } from 'src/pages/UsersList/types';
 import { Actions } from 'src/constants';
 import { BaseUserListModalProps, FormValues } from './types';
-import { createUser, updateUser, atLeastOneRoleOrGroup } from './utils';
+import {
+  createUser,
+  updateUser,
+  atLeastOneRoleOrGroup,
+  handleUserError,
+} from './utils';

 export interface UserModalProps extends BaseUserListModalProps {
  roles: Role[];
@@ -51,36 +56,6 @@ function UserListModal({
 }: UserModalProps) {
  const { addDangerToast, addSuccessToast } = useToasts();
  const handleFormSubmit = async (values: FormValues) => {
-    const handleError = async (
-      err: any,
-      action: Actions.CREATE | Actions.UPDATE,
-    ) => {
-      let errorMessage =
-        action === Actions.CREATE
-          ? t('There was an error creating the user. Please, try again.')
-          : t('There was an error updating the user. Please, try again.');
-
-      if (err.status === 422) {
-        const errorData = await err.json();
-        const detail = errorData?.message || '';
-
-        if (detail.includes('duplicate key value')) {
-          if (detail.includes('ab_user_username_key')) {
-            errorMessage = t(
-              'This username is already taken. Please choose another one.',
-            );
-          } else if (detail.includes('ab_user_email_key')) {
-            errorMessage = t(
-              'This email is already associated with an account. Please choose another one.',
-            );
-          }
-        }
-      }
-
-      addDangerToast(errorMessage);
-      throw err;
-    };
-
    if (isEditMode) {
      if (!user) {
        throw new Error('User is required in edit mode');
@@ -89,14 +64,14 @@ function UserListModal({
        await updateUser(user.id, values);
        addSuccessToast(t('The user has been updated successfully.'));
      } catch (err) {
-        await handleError(err, Actions.UPDATE);
+        await handleUserError(err as Response, Actions.UPDATE, addDangerToast);
      }
    } else {
      try {
        await createUser(values);
        addSuccessToast(t('The user has been created successfully.'));
      } catch (err) {
-        await handleError(err, Actions.CREATE);
+        await handleUserError(err as Response, Actions.CREATE, addDangerToast);
      }
    }
  };
@@ -0,0 +1,99 @@
+/**
+ * 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.
+ */
+import { Actions } from 'src/constants';
+import { handleUserError } from './utils';
+
+test('shows the password validation message from a 400 response', async () => {
+  const error = new Response(
+    JSON.stringify({
+      message: {
+        password: ['Password must be at least 8 characters long.'],
+      },
+    }),
+    { status: 400 },
+  );
+  const addDangerToast = jest.fn();
+
+  await expect(
+    handleUserError(error, Actions.CREATE, addDangerToast),
+  ).rejects.toBe(error);
+  expect(addDangerToast).toHaveBeenCalledWith(
+    'Password must be at least 8 characters long.',
+  );
+});
+
+test('shows a plain string message from a 400 response', async () => {
+  const error = new Response(
+    JSON.stringify({ message: 'User must have at least one role or group!' }),
+    { status: 400 },
+  );
+  const addDangerToast = jest.fn();
+
+  await expect(
+    handleUserError(error, Actions.UPDATE, addDangerToast),
+  ).rejects.toBe(error);
+  expect(addDangerToast).toHaveBeenCalledWith(
+    'User must have at least one role or group!',
+  );
+});
+
+test('keeps the duplicate username message for a 422 response', async () => {
+  const error = new Response(
+    JSON.stringify({
+      message:
+        'duplicate key value violates unique constraint "ab_user_username_key"',
+    }),
+    { status: 422 },
+  );
+  const addDangerToast = jest.fn();
+
+  await expect(
+    handleUserError(error, Actions.CREATE, addDangerToast),
+  ).rejects.toBe(error);
+  expect(addDangerToast).toHaveBeenCalledWith(
+    'This username is already taken. Please choose another one.',
+  );
+});
+
+test('shows the generic message when a 422 response has no message', async () => {
+  const error = new Response(JSON.stringify({ foo: 'bar' }), { status: 422 });
+  const addDangerToast = jest.fn();
+
+  await expect(
+    handleUserError(error, Actions.CREATE, addDangerToast),
+  ).rejects.toBe(error);
+  expect(addDangerToast).toHaveBeenCalledWith(
+    'There was an error creating the user. Please, try again.',
+  );
+});
+
+test('shows the generic message when a 400 response is not JSON', async () => {
+  const error = new Response('<html>Bad request</html>', {
+    status: 400,
+    headers: { 'Content-Type': 'text/html' },
+  });
+  const addDangerToast = jest.fn();
+
+  await expect(
+    handleUserError(error, Actions.CREATE, addDangerToast),
+  ).rejects.toBe(error);
+  expect(addDangerToast).toHaveBeenCalledWith(
+    'There was an error creating the user. Please, try again.',
+  );
+});
@@ -17,10 +17,49 @@
 * under the License.
 */
 import { t } from '@apache-superset/core/translation';
-import { SupersetClient } from '@superset-ui/core';
+import { getClientErrorObject, SupersetClient } from '@superset-ui/core';
 import { SelectOption } from 'src/components/ListView';
+import { Actions } from 'src/constants';
 import { FormValues } from './types';

+type AddDangerToast = (message: string) => void;
+
+export const handleUserError = async (
+  err: Response,
+  action: Actions.CREATE | Actions.UPDATE,
+  addDangerToast: AddDangerToast,
+): Promise<never> => {
+  let errorMessage =
+    action === Actions.CREATE
+      ? t('There was an error creating the user. Please, try again.')
+      : t('There was an error updating the user. Please, try again.');
+
+  if (err.status === 400 || err.status === 422) {
+    const errorData = await getClientErrorObject(err);
+    const message: unknown = errorData.message;
+
+    if (err.status === 400 && message && errorData.error) {
+      errorMessage = errorData.error;
+    } else if (
+      err.status === 422 &&
+      errorData.error?.includes('duplicate key value')
+    ) {
+      if (errorData.error.includes('ab_user_username_key')) {
+        errorMessage = t(
+          'This username is already taken. Please choose another one.',
+        );
+      } else if (errorData.error.includes('ab_user_email_key')) {
+        errorMessage = t(
+          'This email is already associated with an account. Please choose another one.',
+        );
+      }
+    }
+  }
+
+  addDangerToast(errorMessage);
+  throw err;
+};
+
 export const createUser = async (values: FormValues) => {
  const { confirmPassword: _confirmPassword, ...payload } = values;
  if (payload.active == null) {
@@ -97,6 +97,48 @@ describe('apiResource hooks', () => {
        error: fakeError,
      });
    });
+
+    test('skips the fetch and stays loading when skip is true', async () => {
+      const fetchMock = jest.fn().mockResolvedValue(fakeApiResult);
+      (makeApi as jest.Mock).mockReturnValue(fetchMock);
+      const { result } = renderHook(() =>
+        useApiResourceFullBody('/test/endpoint', true),
+      );
+      await act(async () => {
+        jest.runAllTimers();
+      });
+      expect(fetchMock).not.toHaveBeenCalled();
+      expect(result.current).toEqual({
+        status: ResourceStatus.Loading,
+        result: null,
+        error: null,
+      });
+    });
+
+    test('re-enables the fetch when skip toggles from true to false', async () => {
+      const fetchMock = jest.fn().mockResolvedValue(fakeApiResult);
+      (makeApi as jest.Mock).mockReturnValue(fetchMock);
+      const { result, rerender } = renderHook(
+        ({ skip }) => useApiResourceFullBody('/test/endpoint', skip),
+        { initialProps: { skip: true } },
+      );
+      await act(async () => {
+        jest.runAllTimers();
+      });
+      expect(fetchMock).not.toHaveBeenCalled();
+      expect(result.current.status).toEqual(ResourceStatus.Loading);
+
+      rerender({ skip: false });
+      await act(async () => {
+        jest.runAllTimers();
+      });
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+      expect(result.current).toEqual({
+        status: ResourceStatus.Complete,
+        result: fakeApiResult,
+        error: null,
+      });
+    });
  });

  // eslint-disable-next-line no-restricted-globals -- TODO: Migrate from describe blocks
@@ -86,6 +86,7 @@ const initialState: LoadingState = {
 */
 export function useApiResourceFullBody<RESULT>(
  endpoint: string,
+  skip = false,
 ): Resource<RESULT> {
  const [resource, setResource] = useState<Resource<RESULT>>(initialState);
  const cancelRef = useRef<() => void>(() => {});
@@ -98,6 +99,12 @@ export function useApiResourceFullBody<RESULT>(
    // when this effect runs, the endpoint has changed.
    // cancel any current calls so that state doesn't get messed up.
    cancelRef.current();
+
+    // Allow callers to opt out of fetching (e.g. when the identifier isn't
+    // known yet) so we don't fire requests against invalid endpoints.
+    if (skip) {
+      return undefined;
+    }
    let cancelled = false;
    cancelRef.current = () => {
      cancelled = true;
@@ -132,7 +139,7 @@ export function useApiResourceFullBody<RESULT>(
    return () => {
      cancelled = true;
    };
-  }, [endpoint]);
+  }, [endpoint, skip]);

  return resource;
 }
@@ -181,9 +188,12 @@ const extractInnerResult = <T>(responseBody: { result: T }) =>
 *
 * @param endpoint The url where the resource is located.
 */
-export function useApiV1Resource<RESULT>(endpoint: string): Resource<RESULT> {
+export function useApiV1Resource<RESULT>(
+  endpoint: string,
+  skip = false,
+): Resource<RESULT> {
  return useTransformedResource(
-    useApiResourceFullBody<{ result: RESULT }>(endpoint),
+    useApiResourceFullBody<{ result: RESULT }>(endpoint, skip),
    extractInnerResult,
  );
 }
@@ -29,6 +29,7 @@ export {
 export * from './catalogs';
 export * from './charts';
 export * from './dashboards';
+export * from './lineage';
 export * from './tables';
 export * from './schemas';
 export * from './queryValidations';
@@ -0,0 +1,128 @@
+/**
+ * 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.
+ */
+import { renderHook, waitFor } from '@testing-library/react';
+import { makeApi } from '@superset-ui/core';
+import {
+  useChartLineage,
+  useDashboardLineage,
+  useDatasetLineage,
+} from './lineage';
+
+jest.mock('@superset-ui/core', () => ({
+  ...jest.requireActual('@superset-ui/core'),
+  makeApi: jest.fn(),
+}));
+
+const mockedMakeApi = jest.mocked(makeApi);
+
+// makeApi returns a function that issues the request; capture the endpoint it
+// was configured with so we can assert the correct URL was built.
+function mockApiSuccess(payload: unknown) {
+  const fetcher = jest.fn().mockResolvedValue({ result: payload });
+  mockedMakeApi.mockReturnValue(fetcher as any);
+  return fetcher;
+}
+
+function mockApiError(error: Error) {
+  const fetcher = jest.fn().mockRejectedValue(error);
+  mockedMakeApi.mockReturnValue(fetcher as any);
+  return fetcher;
+}
+
+beforeEach(() => {
+  jest.clearAllMocks();
+});
+
+test('useDatasetLineage fetches dataset lineage and unwraps result', async () => {
+  const payload = {
+    dataset: { id: 1, name: 'ds' },
+    upstream: { database: { id: 2, database_name: 'db', backend: 'pg' } },
+    downstream: {
+      charts: { count: 0, result: [] },
+      dashboards: { count: 0, result: [] },
+    },
+  };
+  mockApiSuccess(payload);
+
+  const { result } = renderHook(() => useDatasetLineage(1));
+
+  expect(result.current.status).toBe('loading');
+  await waitFor(() => expect(result.current.status).toBe('complete'));
+
+  expect(mockedMakeApi).toHaveBeenCalledWith(
+    expect.objectContaining({
+      method: 'GET',
+      endpoint: '/api/v1/dataset/1/lineage',
+    }),
+  );
+  expect(result.current.result).toEqual(payload);
+  expect(result.current.error).toBeNull();
+});
+
+test('useChartLineage builds the chart lineage endpoint', async () => {
+  mockApiSuccess({ chart: { id: 5, slice_name: 'c', viz_type: 'pie' } });
+
+  const { result } = renderHook(() => useChartLineage(5));
+
+  await waitFor(() => expect(result.current.status).toBe('complete'));
+  expect(mockedMakeApi).toHaveBeenCalledWith(
+    expect.objectContaining({ endpoint: '/api/v1/chart/5/lineage' }),
+  );
+});
+
+test('useDashboardLineage builds the dashboard lineage endpoint', async () => {
+  mockApiSuccess({ dashboard: { id: 9, title: 'd', slug: 'd' } });
+
+  const { result } = renderHook(() => useDashboardLineage(9));
+
+  await waitFor(() => expect(result.current.status).toBe('complete'));
+  expect(mockedMakeApi).toHaveBeenCalledWith(
+    expect.objectContaining({ endpoint: '/api/v1/dashboard/9/lineage' }),
+  );
+});
+
+test('lineage hooks surface network errors', async () => {
+  mockApiError(new Error('Network error'));
+
+  const { result } = renderHook(() => useDatasetLineage(1));
+
+  await waitFor(() => expect(result.current.status).toBe('error'));
+  expect(result.current.result).toBeNull();
+  expect(result.current.error).toBeInstanceOf(Error);
+});
+
+test('lineage hooks skip the request when the id is empty', async () => {
+  const fetcher = mockApiSuccess({});
+
+  const { result } = renderHook(() => useDatasetLineage(''));
+
+  // Empty id resolves immediately without ever firing a request, so we never
+  // hit an invalid endpoint such as `/api/v1/dataset//lineage`.
+  expect(result.current.status).toBe('loading');
+  expect(fetcher).not.toHaveBeenCalled();
+});
+
+test('lineage hooks skip the request when skip is true', async () => {
+  const fetcher = mockApiSuccess({});
+
+  const { result } = renderHook(() => useChartLineage(5, true));
+
+  expect(result.current.status).toBe('loading');
+  expect(fetcher).not.toHaveBeenCalled();
+});
@@ -0,0 +1,151 @@
+/**
+ * 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.
+ */
+
+import { useApiV1Resource } from './apiResources';
+
+// Database entity type
+export type DatabaseEntity = {
+  id: number;
+  database_name: string;
+  backend: string;
+};
+
+// Dataset entity type
+export type DatasetEntity = {
+  id: number;
+  name: string;
+  schema: string | null;
+  table_name: string;
+  database_id: number;
+  database_name: string;
+  chart_ids?: number[];
+};
+
+// Chart entity type
+export type ChartEntity = {
+  id: number;
+  slice_name: string;
+  viz_type: string;
+  dashboard_ids?: number[];
+  dataset_id?: number;
+};
+
+// Dashboard entity type
+export type DashboardEntity = {
+  id: number;
+  title: string;
+  slug: string;
+  chart_ids?: number[];
+};
+
+// Dataset lineage response type
+export type DatasetLineage = {
+  dataset: DatasetEntity;
+  upstream: {
+    database: DatabaseEntity;
+  };
+  downstream: {
+    charts: {
+      count: number;
+      result: ChartEntity[];
+    };
+    dashboards: {
+      count: number;
+      result: DashboardEntity[];
+    };
+  };
+};
+
+// Chart lineage response type
+export type ChartLineage = {
+  chart: ChartEntity & {
+    datasource_id: number;
+    datasource_type: string;
+  };
+  upstream: {
+    dataset: DatasetEntity;
+    database: DatabaseEntity;
+  };
+  downstream: {
+    dashboards: {
+      count: number;
+      result: DashboardEntity[];
+    };
+  };
+};
+
+// Dashboard lineage response type
+export type DashboardLineage = {
+  dashboard: DashboardEntity & {
+    published: boolean;
+  };
+  upstream: {
+    charts: {
+      count: number;
+      result: ChartEntity[];
+    };
+    datasets: {
+      count: number;
+      result: DatasetEntity[];
+    };
+    databases: {
+      count: number;
+      result: DatabaseEntity[];
+    };
+  };
+  downstream: null;
+};
+
+// A missing/empty identifier means we have nothing to fetch yet; skip the
+// request so we never hit invalid endpoints like `/api/v1/chart//lineage`.
+const isEmptyId = (idOrUuid: string | number): boolean =>
+  idOrUuid === '' || idOrUuid == null;
+
+/**
+ * Hook to fetch lineage data for a dataset
+ * @param idOrUuid Dataset ID or UUID
+ * @param skip When true, defers the request (e.g. until the tab is active)
+ */
+export const useDatasetLineage = (idOrUuid: string | number, skip = false) =>
+  useApiV1Resource<DatasetLineage>(
+    `/api/v1/dataset/${idOrUuid}/lineage`,
+    skip || isEmptyId(idOrUuid),
+  );
+
+/**
+ * Hook to fetch lineage data for a chart
+ * @param idOrUuid Chart ID or UUID
+ * @param skip When true, defers the request (e.g. until the tab is active)
+ */
+export const useChartLineage = (idOrUuid: string | number, skip = false) =>
+  useApiV1Resource<ChartLineage>(
+    `/api/v1/chart/${idOrUuid}/lineage`,
+    skip || isEmptyId(idOrUuid),
+  );
+
+/**
+ * Hook to fetch lineage data for a dashboard
+ * @param idOrSlug Dashboard ID or slug
+ * @param skip When true, defers the request (e.g. until the tab is active)
+ */
+export const useDashboardLineage = (idOrSlug: string | number, skip = false) =>
+  useApiV1Resource<DashboardLineage>(
+    `/api/v1/dashboard/${idOrSlug}/lineage`,
+    skip || isEmptyId(idOrSlug),
+  );
@@ -1157,6 +1157,34 @@ test('dataset links use internal routing when PREVENT_UNSAFE_DEFAULT_URLS_ON_DAT
  });
 });

+test('legacy dashboard default URLs use the registered client route', async () => {
+  const dataset = {
+    ...mockDatasets[0],
+    explore_url: '/superset/dashboard/123/?standalone=1#section',
+  };
+  mockDatasetListEndpoints({ result: [dataset], count: 1 });
+
+  renderDatasetList(
+    mockAdminUser,
+    {},
+    {
+      common: {
+        conf: {
+          PREVENT_UNSAFE_DEFAULT_URLS_ON_DATASET: true,
+        },
+      },
+    },
+  );
+
+  const datasetLink = await screen.findByRole('link', {
+    name: dataset.table_name,
+  });
+  expect(datasetLink).toHaveAttribute(
+    'href',
+    '/dashboard/123/?standalone=1#section',
+  );
+});
+
 // Note: These delete error tests verify that the modal doesn't open when fetching
 // related_objects fails. The component's openDatasetDeleteModal error handler
 // (index.tsx:262-268) returns a string but doesn't call addDangerToast(), so no
@@ -54,10 +54,18 @@ import {

 const APP_ROOT = '/superset';

-const renderUnderSubdirectory = () => {
+const renderUnderSubdirectory = (preventUnsafeDefaultUrls = false) => {
+  const defaultState = createDefaultStoreState(mockAdminUser);
  const store = createMockStore({
-    ...createDefaultStoreState(mockAdminUser),
+    ...defaultState,
    user: mockAdminUser,
+    common: {
+      ...defaultState.common,
+      conf: {
+        ...defaultState.common?.conf,
+        PREVENT_UNSAFE_DEFAULT_URLS_ON_DATASET: preventUnsafeDefaultUrls,
+      },
+    },
  });
  return render(
    <Provider store={store}>
@@ -115,6 +123,31 @@ test('explore link is single-prefixed under a subdirectory deployment', async ()
  expect(exploreLink.getAttribute('href')).not.toContain('/superset/superset');
 });

+test('legacy dashboard default URL uses the router basename once', async () => {
+  // A subdirectory user pastes the full browser path, so the saved value
+  // carries both the application root and the legacy `/superset` prefix.
+  // stripAppRoot removes the root and the legacy normalization removes the
+  // prefix, leaving the basename to re-add the root exactly once.
+  const dataset = {
+    ...mockDatasets[0],
+    explore_url: `${APP_ROOT}/superset/dashboard/123/?standalone=1#section`,
+  };
+  mockDatasetListEndpoints({ result: [dataset], count: 1 });
+
+  renderUnderSubdirectory(true);
+
+  const dashboardLink = await screen.findByRole('link', {
+    name: dataset.table_name,
+  });
+  expect(dashboardLink).toHaveAttribute(
+    'href',
+    `${APP_ROOT}/dashboard/123/?standalone=1#section`,
+  );
+  expect(dashboardLink.getAttribute('href')).not.toContain(
+    '/superset/superset',
+  );
+});
+
 test('external default_endpoint passes through unprefixed', async () => {
  const dataset = {
    ...mockDatasets[0],
@@ -87,7 +87,6 @@ import withToasts from 'src/components/MessageToasts/withToasts';
 import { Icons } from '@superset-ui/core/components/Icons';
 import WarningIconWithTooltip from '@superset-ui/core/components/WarningIconWithTooltip';
 import { isUserEditorOrAdmin } from 'src/dashboard/util/permissionUtils';
-
 import {
  PAGE_SIZE,
  SORT_BY,
@@ -114,6 +113,10 @@ import type {
 } from 'src/types/bootstrapTypes';
 import type User from 'src/types/User';

+// Keep saved Default URLs compatible with the prefix-free SPA route.
+const normalizeLegacyDashboardUrl = (url: string) =>
+  url.replace(/^\/superset(?=\/dashboard(?:\/|$))/, '');
+
 const SEMANTIC_LAYERS_FLAG = 'SEMANTIC_LAYERS' as FeatureFlag;
 type DatasetExtra = {
  certification?: {
@@ -722,7 +725,9 @@ const DatasetList: FunctionComponent<DatasetListProps> = ({
          // Router basename, which re-prefixes the root — so strip it here to
          // avoid a doubled `/superset/superset/...`. External
          // `default_endpoint` URLs pass through unchanged.
-          const exploreTo = stripAppRoot(exploreURL);
+          const exploreTo = normalizeLegacyDashboardUrl(
+            stripAppRoot(exploreURL),
+          );
          let titleLink: JSX.Element;
          if (PREVENT_UNSAFE_DEFAULT_URLS_ON_DATASET) {
            titleLink = (
@@ -28,7 +28,7 @@
        "@typescript-eslint/parser": "^8.67.0",
        "eslint": "^10.8.1",
        "eslint-config-prettier": "^10.1.8",
-        "globals": "^17.9.0",
+        "globals": "^17.10.0",
        "oxfmt": "^0.63.0",
        "tscw-config": "^1.1.2",
        "typescript": "^6.0.3",
@@ -2053,9 +2053,9 @@
      }
    },
    "node_modules/globals": {
-      "version": "17.9.0",
-      "resolved": "https://registry.npmjs.org/globals/-/globals-17.9.0.tgz",
-      "integrity": "sha512-m/MvAW61QVU5VDNF1Vj8axt016h8w7L5TU1e9zlab7XIttAT2YAlCwl75K1fOqvMM9apmD7lbCIRhpfkhmxhCg==",
+      "version": "17.10.0",
+      "resolved": "https://registry.npmjs.org/globals/-/globals-17.10.0.tgz",
+      "integrity": "sha512-V0kztuWST2k8A/VbxAY8+L+7+Rgo3fyA24IHRLrZp7HOzJjV0gHSaZUjK9lpP/IrBSNite2tZ1prhRkinRu1CA==",
      "dev": true,
      "license": "MIT",
      "engines": {
@@ -36,7 +36,7 @@
    "@typescript-eslint/parser": "^8.67.0",
    "eslint": "^10.8.1",
    "eslint-config-prettier": "^10.1.8",
-    "globals": "^17.9.0",
+    "globals": "^17.10.0",
    "oxfmt": "^0.63.0",
    "tscw-config": "^1.1.2",
    "typescript": "^6.0.3",
@@ -172,6 +172,7 @@ class ChartRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
        "get_version",
        "activity",
        "restore_version",
+        "lineage",
    }
    class_permission_name = "Chart"
    # Custom methods (``restore``) need an explicit entry; FAB's @protect()
@@ -515,6 +516,114 @@ class ChartRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
        ]
        return self.response(200, result=result)

+    @expose("/<id_or_uuid>/lineage", methods=("GET",))
+    @protect()
+    @safe
+    @statsd_metrics
+    @event_logger.log_this_with_context(
+        action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.lineage",
+        log_to_statsd=False,
+    )
+    def lineage(self, id_or_uuid: str) -> Response:
+        """Get lineage information for a chart.
+        ---
+        get:
+          summary: Get lineage information for a chart
+          description: >-
+            Returns upstream (dataset, database) and downstream (dashboards) lineage
+            information for a chart
+          parameters:
+          - in: path
+            name: id_or_uuid
+            schema:
+              type: string
+            description: Either the id of the chart, or its uuid
+          responses:
+            200:
+              description: Lineage information
+              content:
+                application/json:
+                  schema:
+                    $ref: "#/components/schemas/ChartLineageResponseSchema"
+            401:
+              $ref: '#/components/responses/401'
+            404:
+              $ref: '#/components/responses/404'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        try:
+            chart = ChartDAO.get_by_id_or_uuid(id_or_uuid)
+        except ChartNotFoundError:
+            return self.response_404()
+
+        chart_info = {
+            "id": chart.id,
+            "slice_name": chart.slice_name,
+            "viz_type": chart.viz_type,
+        }
+
+        # Get upstream (dataset and database) information. Schema/table/database
+        # details are only exposed to users who can access the underlying
+        # datasource; otherwise they are redacted so lineage never leaks
+        # datasource internals (the dataset id/name are kept so the graph
+        # still renders), mirroring the dashboard lineage endpoint.
+        upstream: dict[str, Any] = {}
+        if dataset := chart.datasource:
+            can_access = security_manager.can_access_datasource(dataset)
+            upstream["dataset"] = {
+                "id": dataset.id,
+                "name": dataset.name,
+                "database_id": dataset.database_id if can_access else None,
+                "database_name": (
+                    dataset.database.database_name
+                    if can_access and dataset.database
+                    else None
+                ),
+                "schema": dataset.schema if can_access else None,
+                "table_name": dataset.table_name if can_access else None,
+            }
+            if can_access and dataset.database:
+                upstream["database"] = {
+                    "id": dataset.database.id,
+                    "database_name": dataset.database.database_name,
+                    "backend": dataset.database.backend,
+                }
+            else:
+                upstream["database"] = None
+        else:
+            upstream["dataset"] = None
+            upstream["database"] = None
+
+        # Get downstream (dashboards) information, filtered by the current
+        # user's permissions so lineage never exposes dashboards the user
+        # cannot access.
+        dashboards: list[dict[str, Any]] = []
+        for dashboard in chart.dashboards:
+            if not security_manager.can_access_dashboard(dashboard):
+                continue
+            dashboards.append(
+                {
+                    "id": dashboard.id,
+                    "title": dashboard.dashboard_title,
+                    "slug": dashboard.slug,
+                }
+            )
+
+        downstream = {
+            "dashboards": {
+                "count": len(dashboards),
+                "result": dashboards,
+            },
+        }
+
+        result = {
+            "chart": chart_info,
+            "upstream": upstream,
+            "downstream": downstream,
+        }
+        return self.response(200, result=result)
+
    @expose("/", methods=("POST",))
    @protect()
    @safe
@@ -1899,6 +1899,53 @@ class ChartGetResponseSchema(Schema):
    datasource_uuid = fields.UUID(attribute="table.uuid")


+class ChartLineageChartSchema(Schema):
+    id = fields.Integer()
+    slice_name = fields.String()
+    viz_type = fields.String()
+
+
+class ChartLineageDatasetSchema(Schema):
+    id = fields.Integer()
+    name = fields.String()
+    database_id = fields.Integer()
+    database_name = fields.String(allow_none=True)
+    schema = fields.String(allow_none=True)
+    table_name = fields.String()
+
+
+class ChartLineageDatabaseSchema(Schema):
+    id = fields.Integer()
+    database_name = fields.String()
+    backend = fields.String()
+
+
+class ChartLineageDashboardSchema(Schema):
+    id = fields.Integer()
+    title = fields.String()
+    slug = fields.String(allow_none=True)
+
+
+class ChartLineageUpstreamSchema(Schema):
+    dataset = fields.Nested(ChartLineageDatasetSchema, allow_none=True)
+    database = fields.Nested(ChartLineageDatabaseSchema, allow_none=True)
+
+
+class ChartLineageDownstreamDashboardsSchema(Schema):
+    count = fields.Integer()
+    result = fields.List(fields.Nested(ChartLineageDashboardSchema))
+
+
+class ChartLineageDownstreamSchema(Schema):
+    dashboards = fields.Nested(ChartLineageDownstreamDashboardsSchema)
+
+
+class ChartLineageResponseSchema(Schema):
+    chart = fields.Nested(ChartLineageChartSchema)
+    upstream = fields.Nested(ChartLineageUpstreamSchema)
+    downstream = fields.Nested(ChartLineageDownstreamSchema)
+
+
 CHART_SCHEMAS = (
    ChartCacheWarmUpRequestSchema,
    ChartCacheWarmUpResponseSchema,
@@ -1926,4 +1973,5 @@ CHART_SCHEMAS = (
    ChartGetResponseSchema,
    ChartCacheScreenshotResponseSchema,
    GetFavStarIdsSchema,
+    ChartLineageResponseSchema,
 )
@@ -17,6 +17,7 @@
 # pylint: disable=invalid-name
 from __future__ import annotations

+import inspect
 import logging
 from datetime import datetime
 from pprint import pformat
@@ -205,8 +206,86 @@ class QueryObject:  # pylint: disable=too-many-instance-attributes
    def _set_post_processing(
        self, post_processing: list[dict[str, Any] | None] | None
    ) -> None:
-        post_processing = post_processing or []
-        self.post_processing = [post_proc for post_proc in post_processing if post_proc]
+        self.post_processing = [
+            self._drop_unsupported_options(post_proc)
+            for post_proc in post_processing or []
+            if post_proc
+        ]
+
+    @staticmethod
+    def _drop_unsupported_options(post_proc: dict[str, Any]) -> dict[str, Any]:
+        """
+        Drop options that the post-processing operation no longer accepts.
+
+        A chart's ``query_context`` is written when the chart is saved and is
+        never rewritten afterwards, while Explore rebuilds the query from
+        ``form_data`` at every render. A chart saved by an older version of
+        Superset can therefore reference an option that has since been removed
+        from the operation. ``exec_post_processing`` passes the stored options
+        as keyword arguments, so that option raises a bare ``TypeError`` on
+        every path that replays the stored ``query_context`` -- the chart data
+        endpoint, alerts and reports, thumbnails, CSV export -- while the same
+        chart still renders correctly in Explore.
+
+        Comparing against the signature avoids a hard-coded list of removed
+        option names, which would need extending at each release.
+        """
+        operation = post_proc.get("operation")
+        function = (
+            getattr(pandas_postprocessing, operation, None)
+            if isinstance(operation, str)
+            else None
+        )
+        if function is None:
+            # A missing or unknown operation is left untouched, so that
+            # exec_post_processing reports it as InvalidPostProcessingError.
+            return post_proc
+
+        parameters = inspect.signature(function).parameters
+        if any(
+            parameter.kind is inspect.Parameter.VAR_KEYWORD
+            for parameter in parameters.values()
+        ):
+            return post_proc
+
+        # `exec_post_processing` calls the operation as `operation(df, **options)`,
+        # so an option can only reach a parameter that a caller may fill by
+        # keyword. That excludes the first parameter, which receives the
+        # DataFrame positionally, and any positional-only or `*args` parameter.
+        keyword_parameters = {
+            name
+            for position, (name, parameter) in enumerate(parameters.items())
+            if position > 0
+            and parameter.kind
+            in (
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                inspect.Parameter.KEYWORD_ONLY,
+            )
+        }
+
+        options = post_proc.get("options") or {}
+        unsupported = {key for key in options if key not in keyword_parameters}
+        if not unsupported:
+            return post_proc
+
+        # Logged at info: a chart saved before the option was removed hits this
+        # on every render, so a warning would repeat for as long as the chart
+        # is not resaved, without anything new to report.
+        logger.info(
+            "Dropping unsupported option(s) %s of post-processing operation "
+            "`%s`. The chart's stored query_context predates the current "
+            "signature of that operation.",
+            sorted(unsupported),
+            operation,
+        )
+        return {
+            **post_proc,
+            "options": {
+                key: value
+                for key, value in options.items()
+                if key in keyword_parameters
+            },
+        }

    def _init_series_columns(
        self,
@@ -957,7 +957,13 @@ class AnnotationDatasource(BaseDatasource):
    def get_query_str(self, query_obj: QueryObjectDict) -> str:
        raise NotImplementedError()

-    def values_for_column(self, column_name: str, limit: int = 10000) -> list[Any]:
+    def values_for_column(
+        self,
+        column_name: str,
+        limit: int = 10000,
+        denormalize_column: bool = False,
+        array_elements: bool = False,
+    ) -> list[Any]:
        raise NotImplementedError()


@@ -182,6 +182,7 @@ MODEL_API_RW_METHOD_PERMISSION_MAP = {
    "list_versions": "read",
    "get_version": "read",
    "activity": "read",
+    "lineage": "read",
 }

 EXTRA_FORM_DATA_APPEND_KEYS = {
@@ -23,7 +23,7 @@ from typing import Any, Dict, List
 import dateutil.parser
 from sqlalchemy import or_, select
 from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy.orm import joinedload, Query
+from sqlalchemy.orm import joinedload, Query, selectinload

 from superset.connectors.sqla.models import (
    RLSFilterTables,
@@ -149,6 +149,9 @@ class DatasetDAO(BaseDAO[SqlaTable]):
                Slice.datasource_id == database_id,
                Slice.datasource_type == DatasourceType.TABLE,
            )
+            # Eager-load the related dashboards so callers (e.g. lineage) can
+            # iterate ``chart.dashboards`` without triggering a query per chart.
+            .options(selectinload(Slice.dashboards))
            .all()
        )
        chart_ids = [chart.id for chart in charts]
@@ -158,6 +161,9 @@ class DatasetDAO(BaseDAO[SqlaTable]):
                db.session.query(Dashboard)
                .join(Dashboard.slices)
                .filter(Slice.id.in_(chart_ids))
+                # Eager-load slices so callers can iterate ``dashboard.slices``
+                # without a query per dashboard.
+                .options(selectinload(Dashboard.slices))
            )
            .distinct()
            .all()
@@ -116,6 +116,7 @@ from superset.dashboards.schemas import (
    DashboardExportXlsxPostSchema,
    DashboardExportXlsxResponseSchema,
    DashboardGetResponseSchema,
+    DashboardLineageResponseSchema,
    DashboardNativeFiltersConfigUpdateSchema,
    DashboardPostSchema,
    DashboardPutSchema,
@@ -329,6 +330,7 @@ class DashboardRestApi(
        "get_version",
        "activity",
        "restore_version",
+        "lineage",
    }
    resource_name = "dashboard"
    allow_browser_login = True
@@ -562,9 +564,11 @@ class DashboardRestApi(
        DashboardCacheScreenshotResponseSchema,
        DashboardCopySchema,
        DashboardGetResponseSchema,
+        DashboardLineageResponseSchema,
        DashboardDatasetSchema,
        DashboardExportXlsxPostSchema,
        DashboardExportXlsxResponseSchema,
+        DashboardLineageResponseSchema,
        TabsPayloadSchema,
        GetFavStarIdsSchema,
        EmbeddedDashboardResponseSchema,
@@ -680,6 +684,126 @@ class DashboardRestApi(
            current_entity_etag_uuid(Dashboard, dash.id, dash.uuid),
        )

+    @expose("/<id_or_slug>/lineage", methods=("GET",))
+    @protect()
+    @safe
+    @statsd_metrics
+    @with_dashboard
+    @event_logger.log_this_with_context(
+        action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.lineage",
+        log_to_statsd=False,
+    )
+    # pylint: disable=arguments-differ,arguments-renamed
+    def lineage(self, dash: Dashboard) -> Response:
+        """Get lineage information for a dashboard.
+        ---
+        get:
+          summary: Get lineage information for a dashboard
+          description: >-
+            Returns upstream (charts, datasets, databases) lineage information
+            for a dashboard
+          parameters:
+          - in: path
+            name: id_or_slug
+            schema:
+              type: string
+            description: Either the id of the dashboard, or its slug
+          responses:
+            200:
+              description: Lineage information
+              content:
+                application/json:
+                  schema:
+                    $ref: "#/components/schemas/DashboardLineageResponseSchema"
+            401:
+              $ref: '#/components/responses/401'
+            404:
+              $ref: '#/components/responses/404'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        dashboard_info = {
+            "id": dash.id,
+            "title": dash.dashboard_title,
+            "slug": dash.slug,
+            "published": dash.published,
+        }
+
+        # Get upstream (charts, datasets, databases) information
+        charts: list[dict[str, Any]] = []
+        dataset_map: dict[int, dict[str, Any]] = {}
+        database_map: dict[int, dict[str, Any]] = {}
+
+        for chart in dash.slices:
+            charts.append(
+                {
+                    "id": chart.id,
+                    "slice_name": chart.slice_name,
+                    "viz_type": chart.viz_type,
+                    "dataset_id": chart.datasource_id,
+                }
+            )
+
+            # Collect dataset information. Schema/table/database details are
+            # only exposed to users who can access the underlying datasource;
+            # otherwise they are redacted so lineage never leaks datasource
+            # internals (the dataset id/name are kept so the graph still
+            # renders).
+            dataset = chart.datasource
+            if dataset and dataset.id not in dataset_map:
+                can_access = security_manager.can_access_datasource(dataset)
+                dataset_map[dataset.id] = {
+                    "id": dataset.id,
+                    "name": dataset.name,
+                    "database_id": dataset.database_id if can_access else None,
+                    "database_name": (
+                        dataset.database.database_name
+                        if can_access and dataset.database
+                        else None
+                    ),
+                    "schema": dataset.schema if can_access else None,
+                    "table_name": dataset.table_name if can_access else None,
+                    "chart_ids": [],
+                }
+
+            if dataset and dataset.id in dataset_map:
+                dataset_map[dataset.id]["chart_ids"].append(chart.id)
+
+            # Collect database information, only for accessible datasources
+            if (
+                dataset
+                and security_manager.can_access_datasource(dataset)
+                and dataset.database
+                and dataset.database.id not in database_map
+            ):
+                database_map[dataset.database.id] = {
+                    "id": dataset.database.id,
+                    "database_name": dataset.database.database_name,
+                    "backend": dataset.database.backend,
+                }
+
+        upstream = {
+            "charts": {
+                "count": len(charts),
+                "result": charts,
+            },
+            "datasets": {
+                "count": len(dataset_map),
+                "result": list(dataset_map.values()),
+            },
+            "databases": {
+                "count": len(database_map),
+                "result": list(database_map.values()),
+            },
+        }
+
+        result = {
+            "dashboard": dashboard_info,
+            "upstream": upstream,
+            "downstream": None,
+        }
+        return self.response(200, result=result)
+
    @expose("/<id_or_slug>/datasets", methods=("GET",))
    @protect()
    @handle_api_exception
@@ -668,3 +668,62 @@ class DashboardExportXlsxResponseSchema(Schema):
    job_id = fields.String(
        metadata={"description": "Correlation id for the async export task"}
    )
+
+
+class DashboardLineageDashboardSchema(Schema):
+    id = fields.Integer()
+    title = fields.String()
+    slug = fields.String(allow_none=True)
+    published = fields.Boolean()
+
+
+class DashboardLineageChartSchema(Schema):
+    id = fields.Integer()
+    slice_name = fields.String()
+    viz_type = fields.String()
+    dataset_id = fields.Integer()
+
+
+class DashboardLineageDatasetSchema(Schema):
+    id = fields.Integer()
+    name = fields.String()
+    # database/schema/table details are redacted to ``None`` when the user
+    # cannot access the underlying datasource, so they must be nullable.
+    database_id = fields.Integer(allow_none=True)
+    database_name = fields.String(allow_none=True)
+    schema = fields.String(allow_none=True)
+    table_name = fields.String(allow_none=True)
+    chart_ids = fields.List(fields.Integer())
+
+
+class DashboardLineageDatabaseSchema(Schema):
+    id = fields.Integer()
+    database_name = fields.String()
+    backend = fields.String()
+
+
+class DashboardLineageUpstreamChartsSchema(Schema):
+    count = fields.Integer()
+    result = fields.List(fields.Nested(DashboardLineageChartSchema))
+
+
+class DashboardLineageUpstreamDatasetsSchema(Schema):
+    count = fields.Integer()
+    result = fields.List(fields.Nested(DashboardLineageDatasetSchema))
+
+
+class DashboardLineageUpstreamDatabasesSchema(Schema):
+    count = fields.Integer()
+    result = fields.List(fields.Nested(DashboardLineageDatabaseSchema))
+
+
+class DashboardLineageUpstreamSchema(Schema):
+    charts = fields.Nested(DashboardLineageUpstreamChartsSchema)
+    datasets = fields.Nested(DashboardLineageUpstreamDatasetsSchema)
+    databases = fields.Nested(DashboardLineageUpstreamDatabasesSchema)
+
+
+class DashboardLineageResponseSchema(Schema):
+    dashboard = fields.Nested(DashboardLineageDashboardSchema)
+    upstream = fields.Nested(DashboardLineageUpstreamSchema)
+    downstream = fields.Field(allow_none=True)
@@ -77,6 +77,7 @@ from superset.datasets.schemas import (
    DatasetCacheWarmUpResponseSchema,
    DatasetDrillInfoSchema,
    DatasetDuplicateSchema,
+    DatasetLineageResponseSchema,
    DatasetPostSchema,
    DatasetPutSchema,
    DatasetRelatedObjectsResponse,
@@ -166,6 +167,7 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
        "get_version",
        "activity",
        "restore_version",
+        "lineage",
    }
    list_columns = [
        "id",
@@ -384,6 +386,7 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
        DatasetDuplicateSchema,
        GetOrCreateDatasetSchema,
        VersionListItemSchema,
+        DatasetLineageResponseSchema,
    )

    openapi_spec_methods = openapi_spec_methods_override
@@ -1095,6 +1098,130 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
            dashboards={"count": len(dashboards), "result": dashboards},
        )

+    @expose("/<id_or_uuid>/lineage", methods=("GET",))
+    @protect()
+    @safe
+    @statsd_metrics
+    @event_logger.log_this_with_context(
+        action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.lineage",
+        log_to_statsd=False,
+    )
+    def lineage(self, id_or_uuid: str) -> Response:
+        """Get lineage information for a dataset.
+        ---
+        get:
+          summary: Get lineage information for a dataset
+          description: >-
+            Returns upstream (database) and downstream (charts, dashboards) lineage
+            information for a dataset
+          parameters:
+          - in: path
+            name: id_or_uuid
+            schema:
+              type: string
+            description: Either the id of the dataset, or its uuid
+          responses:
+            200:
+              description: Lineage information
+              content:
+                application/json:
+                  schema:
+                    $ref: "#/components/schemas/DatasetLineageResponseSchema"
+            401:
+              $ref: '#/components/responses/401'
+            404:
+              $ref: '#/components/responses/404'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        dataset = DatasetDAO.find_by_id_or_uuid(id_or_uuid)
+        if not dataset:
+            return self.response_404()
+
+        dataset_info: dict[str, Any] = {
+            "id": dataset.id,
+            "name": dataset.name,
+            "database_id": dataset.database_id,
+            "database_name": (
+                dataset.database.database_name if dataset.database else None
+            ),
+            "schema": dataset.schema,
+            "table_name": dataset.table_name,
+        }
+
+        # Get upstream (database) information
+        upstream: dict[str, Any] = {}
+        if dataset.database:
+            upstream["database"] = {
+                "id": dataset.database.id,
+                "database_name": dataset.database.database_name,
+                "backend": dataset.database.backend,
+            }
+        else:
+            upstream["database"] = None
+
+        # Get downstream (charts and dashboards) information
+        related_data: dict[str, Any] = DatasetDAO.get_related_objects(dataset.id)
+
+        # Build chart information with dashboard IDs, filtering both the charts
+        # and their linked dashboards by the current user's permissions so
+        # lineage never exposes assets the user cannot access.
+        charts: list[dict[str, Any]] = []
+        for chart in related_data["charts"]:
+            if not security_manager.can_access_chart(chart):
+                continue
+            dashboard_ids: list[int] = [
+                d.id
+                for d in chart.dashboards
+                if security_manager.can_access_dashboard(d)
+            ]
+            charts.append(
+                {
+                    "id": chart.id,
+                    "slice_name": chart.slice_name,
+                    "viz_type": chart.viz_type,
+                    "dashboard_ids": dashboard_ids,
+                }
+            )
+
+        # Build dashboard information with chart IDs
+        dashboards: list[dict[str, Any]] = []
+        for dashboard in related_data["dashboards"]:
+            if not security_manager.can_access_dashboard(dashboard):
+                continue
+            chart_ids: list[int] = [
+                chart.id
+                for chart in dashboard.slices
+                if chart.datasource_id == dataset.id
+                and security_manager.can_access_chart(chart)
+            ]
+            dashboards.append(
+                {
+                    "id": dashboard.id,
+                    "title": dashboard.dashboard_title,
+                    "slug": dashboard.slug,
+                    "chart_ids": chart_ids,
+                }
+            )
+
+        downstream: dict[str, Any] = {
+            "charts": {
+                "count": len(charts),
+                "result": charts,
+            },
+            "dashboards": {
+                "count": len(dashboards),
+                "result": dashboards,
+            },
+        }
+
+        result: dict[str, Any] = {
+            "dataset": dataset_info,
+            "upstream": upstream,
+            "downstream": downstream,
+        }
+        return self.response(200, result=result)
+
    @expose("/", methods=("DELETE",))
    @protect()
    @safe
@@ -260,6 +260,60 @@ class DatasetRelatedObjectsResponse(Schema):
    dashboards = fields.Nested(DatasetRelatedDashboards)


+class DatasetLineageDatasetSchema(Schema):
+    id = fields.Integer()
+    name = fields.String()
+    database_id = fields.Integer()
+    database_name = fields.String(allow_none=True)
+    schema = fields.String(allow_none=True)
+    table_name = fields.String()
+
+
+class DatasetLineageDatabaseSchema(Schema):
+    id = fields.Integer()
+    database_name = fields.String()
+    backend = fields.String()
+
+
+class DatasetLineageChartSchema(Schema):
+    id = fields.Integer()
+    slice_name = fields.String()
+    viz_type = fields.String()
+    dashboard_ids = fields.List(fields.Integer())
+
+
+class DatasetLineageDashboardSchema(Schema):
+    id = fields.Integer()
+    title = fields.String()
+    slug = fields.String(allow_none=True)
+    chart_ids = fields.List(fields.Integer())
+
+
+class DatasetLineageUpstreamSchema(Schema):
+    database = fields.Nested(DatasetLineageDatabaseSchema, allow_none=True)
+
+
+class DatasetLineageDownstreamChartsSchema(Schema):
+    count = fields.Integer()
+    result = fields.List(fields.Nested(DatasetLineageChartSchema))
+
+
+class DatasetLineageDownstreamDashboardsSchema(Schema):
+    count = fields.Integer()
+    result = fields.List(fields.Nested(DatasetLineageDashboardSchema))
+
+
+class DatasetLineageDownstreamSchema(Schema):
+    charts = fields.Nested(DatasetLineageDownstreamChartsSchema)
+    dashboards = fields.Nested(DatasetLineageDownstreamDashboardsSchema)
+
+
+class DatasetLineageResponseSchema(Schema):
+    dataset = fields.Nested(DatasetLineageDatasetSchema)
+    upstream = fields.Nested(DatasetLineageUpstreamSchema)
+    downstream = fields.Nested(DatasetLineageDownstreamSchema)
+
+
 class ImportV1ColumnSchema(Schema):
    # pylint: disable=unused-argument
    @pre_load
@@ -133,6 +133,9 @@ class DatasourceRestApi(BaseSupersetApi):

        row_limit = apply_max_row_limit(app.config["FILTER_SELECT_ROW_LIMIT"])
        denormalize_column = not datasource.normalize_columns
+        # Element-level operators (Contains any / Contains all) request the
+        # distinct array *elements* rather than distinct whole arrays.
+        array_elements = parse_boolean_string(request.args.get("array_elements"))

        # Cache distinct column-value results so a dashboard with many filters
        # backed by the same (often heavy) virtual dataset doesn't re-execute
@@ -165,6 +168,7 @@ class DatasourceRestApi(BaseSupersetApi):
                        "col": column_name,
                        "limit": row_limit,
                        "denorm": denormalize_column,
+                        "elements": array_elements,
                        "rls": security_manager.get_rls_cache_key(datasource),
                        "changed_on": str(getattr(datasource, "changed_on", "")),
                    },
@@ -189,6 +193,7 @@ class DatasourceRestApi(BaseSupersetApi):
                column_name=column_name,
                limit=row_limit,
                denormalize_column=denormalize_column,
+                array_elements=array_elements,
            )
        except KeyError:
            return self.response(
@@ -55,7 +55,13 @@ from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.url import URL
 from sqlalchemy.ext.compiler import compiles
 from sqlalchemy.sql import literal_column, quoted_name, text
-from sqlalchemy.sql.expression import BinaryExpression, ColumnClause, Select, TextClause
+from sqlalchemy.sql.expression import (
+    BinaryExpression,
+    ColumnClause,
+    ColumnElement,
+    Select,
+    TextClause,
+)
 from sqlalchemy.types import TypeEngine

 from superset import db
@@ -528,6 +534,11 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
    time_groupby_inline = False
    limit_method = LimitMethod.FORCE_LIMIT
    supports_multivalues_insert = False
+    # Whether this engine supports first-class multi-value (array-typed) columns.
+    # When True, array columns are classified as ``GenericDataType.MULTI_VALUE`` and
+    # the ``array_*`` capability methods below must be implemented. Defaults to
+    # False so engines that have not opted in keep treating arrays as strings.
+    supports_multivalue_columns = False
    allows_joins = True
    allows_subqueries = True
    allows_alias_in_select = True
@@ -2571,6 +2582,105 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
            logger.error(ex, exc_info=True)
            raise

+    @classmethod
+    def array_contains_any(cls, col: ColumnElement, values: list[Any]) -> ColumnElement:
+        """
+        Build a boolean expression testing whether array column ``col`` contains
+        **any** of ``values`` (element-level membership, like ``IN``). Engines
+        that set ``supports_multivalue_columns = True`` must override this with
+        their native function (e.g. ClickHouse ``hasAny``).
+
+        :param col: SQLAlchemy column element for the array column
+        :param values: element values to look for inside the array
+        :return: a SQLAlchemy boolean expression
+        """
+        raise NotImplementedError(
+            f"{cls.engine} does not support multi-value (array) columns"
+        )
+
+    @classmethod
+    def array_contains_all(cls, col: ColumnElement, values: list[Any]) -> ColumnElement:
+        """
+        Build a boolean expression testing whether array column ``col`` contains
+        **all** of ``values``. Engines that set
+        ``supports_multivalue_columns = True`` must override this with their
+        native function (e.g. ClickHouse ``hasAll``).
+
+        :param col: SQLAlchemy column element for the array column
+        :param values: element values that must all be present
+        :return: a SQLAlchemy boolean expression
+        """
+        raise NotImplementedError(
+            f"{cls.engine} does not support multi-value (array) columns"
+        )
+
+    @classmethod
+    def array_length(cls, col: ColumnElement) -> ColumnElement:
+        """
+        Build a numeric expression returning the number of elements in array
+        column ``col``. Engines that set ``supports_multivalue_columns = True``
+        must override this with their native array-length function. Used both for
+        the ``Length`` filter and the ``Is empty`` / ``Is not empty`` operators.
+
+        :param col: SQLAlchemy column element for the array column
+        :return: a SQLAlchemy numeric expression
+        """
+        raise NotImplementedError(
+            f"{cls.engine} does not support multi-value (array) columns"
+        )
+
+    @classmethod
+    def array_literal(cls, values: list[Any]) -> ColumnElement:
+        """
+        Build an array-literal expression from ``values`` (e.g. ClickHouse
+        ``array(v1, v2)`` == ``[v1, v2]``). Used for the whole-array (column-
+        level) operators ``=`` / ``!=`` / ``IN`` / ``NOT IN`` where the array is
+        compared as a single value. Engines that set
+        ``supports_multivalue_columns = True`` must override this.
+
+        :param values: element values that make up the array
+        :return: a SQLAlchemy array-literal expression
+        """
+        raise NotImplementedError(
+            f"{cls.engine} does not support multi-value (array) columns"
+        )
+
+    @classmethod
+    def array_explode(cls, col: ColumnElement) -> ColumnElement:
+        """
+        Build an expression that expands array column ``col`` into one row per
+        element (e.g. ClickHouse ``arrayJoin``). Used to source **element-level**
+        value suggestions (``SELECT DISTINCT array_explode(col)``) for the
+        ``Contains any`` / ``Contains all`` filter operators, so the picker offers
+        individual elements rather than whole arrays. Engines that set
+        ``supports_multivalue_columns = True`` must override this.
+
+        :param col: SQLAlchemy column element for the array column
+        :return: a SQLAlchemy expression yielding one element per row
+        """
+        raise NotImplementedError(
+            f"{cls.engine} does not support multi-value (array) columns"
+        )
+
+    @classmethod
+    def get_array_element_type(  # pylint: disable=unused-argument
+        cls, native_type: str | None
+    ) -> GenericDataType | None:
+        """
+        Return the generic type of an array column's **element** type, derived
+        from its native type string (e.g. ClickHouse ``Array(Int32)`` ->
+        ``NUMERIC``), or ``None`` when the engine has no array support or the
+        element type cannot be resolved.
+
+        Callers use this to coerce filter values to the element type before
+        building array expressions, so, for example, a ``Contains any`` filter on
+        a numeric array compares against numbers rather than quoted strings.
+
+        :param native_type: native column type string of the array column
+        :return: the element's :class:`GenericDataType`, or ``None``
+        """
+        return None
+
    @classmethod
    def get_column_spec(  # pylint: disable=unused-argument
        cls,
@@ -26,8 +26,9 @@ from flask import current_app as app
 from flask_babel import gettext as __
 from marshmallow import fields, Schema
 from marshmallow.validate import Range
-from sqlalchemy import types
+from sqlalchemy import func, types
 from sqlalchemy.engine.url import URL
+from sqlalchemy.sql.expression import ColumnElement
 from urllib3.exceptions import NewConnectionError

 from superset.databases.utils import make_url_safe
@@ -55,6 +56,7 @@ class ClickHouseBaseEngineSpec(BaseEngineSpec):

    time_groupby_inline = True
    supports_multivalues_insert = True
+    supports_multivalue_columns = True

    # ClickHouse doesn't support IS true/false syntax, use = true/false instead
    use_equality_for_boolean_filters = True
@@ -128,12 +130,18 @@ class ClickHouseBaseEngineSpec(BaseEngineSpec):

    column_type_mappings = (
        (
-            re.compile(r".*Enum.*", re.IGNORECASE),
+            # Anchor to the start so only top-level arrays match. This must be
+            # ordered before the ``Enum`` entry below: ``Array(Enum8(...))`` is a
+            # real array and should classify as MULTI_VALUE, not STRING. The
+            # anchor also prevents over-matching nested arrays such as
+            # ``Map(String, Array(String))`` or ``Tuple(Array(String))``, which
+            # are not themselves array columns and must keep their own type.
+            re.compile(r"^Array\(", re.IGNORECASE),
            types.String(),
-            GenericDataType.STRING,
+            GenericDataType.MULTI_VALUE,
        ),
        (
-            re.compile(r".*Array.*", re.IGNORECASE),
+            re.compile(r".*Enum.*", re.IGNORECASE),
            types.String(),
            GenericDataType.STRING,
        ),
@@ -174,6 +182,56 @@ class ClickHouseBaseEngineSpec(BaseEngineSpec):
        ),
    )

+    @classmethod
+    def array_contains_any(cls, col: ColumnElement, values: list[Any]) -> ColumnElement:
+        # ClickHouse: hasAny(arr, [v1, v2]) -> 1 if arr shares any element.
+        # func.array(*values) renders as array(v1, v2) == [v1, v2].
+        return func.hasAny(col, func.array(*values))
+
+    @classmethod
+    def array_contains_all(cls, col: ColumnElement, values: list[Any]) -> ColumnElement:
+        # ClickHouse: hasAll(arr, [v1, v2]) -> 1 if arr contains all elements.
+        return func.hasAll(col, func.array(*values))
+
+    @classmethod
+    def array_length(cls, col: ColumnElement) -> ColumnElement:
+        # ClickHouse: length(arr) -> number of elements
+        return func.length(col)
+
+    @classmethod
+    def array_literal(cls, values: list[Any]) -> ColumnElement:
+        # ClickHouse: array(v1, v2) is equivalent to the literal [v1, v2].
+        return func.array(*values)
+
+    @classmethod
+    def array_explode(cls, col: ColumnElement) -> ColumnElement:
+        # ClickHouse: arrayJoin(arr) yields one row per element, so
+        # SELECT DISTINCT arrayJoin(arr) returns the distinct elements.
+        return func.arrayJoin(col)
+
+    # Matches the element type inside a top-level ``Array(...)`` column, e.g.
+    # ``Array(Int32)`` -> ``Int32``, ``Array(Nullable(String))`` -> ``String``.
+    _ARRAY_ELEMENT_RE = re.compile(r"^Array\((?P<inner>.+)\)$", re.IGNORECASE)
+    # Element-type wrappers that don't change the underlying generic type.
+    _ELEMENT_WRAPPER_RE = re.compile(
+        r"^(?:Nullable|LowCardinality)\((?P<inner>.+)\)$", re.IGNORECASE
+    )
+
+    @classmethod
+    def get_array_element_type(cls, native_type: str | None) -> GenericDataType | None:
+        if not native_type:
+            return None
+        match = cls._ARRAY_ELEMENT_RE.match(native_type.strip())
+        if not match:
+            return None
+        inner = match.group("inner").strip()
+        # Peel wrappers (Nullable/LowCardinality) that don't alter the generic
+        # type so the inner scalar type drives classification.
+        while wrapper := cls._ELEMENT_WRAPPER_RE.match(inner):
+            inner = wrapper.group("inner").strip()
+        spec = cls.get_column_spec(inner)
+        return spec.generic_type if spec else None
+
    @classmethod
    def epoch_to_dttm(cls) -> str:
        return "{col}"
@@ -63,6 +63,7 @@ from superset.mcp_service.utils import (
    sanitize_for_llm_context,
 )
 from superset.mcp_service.utils.response_utils import humanize_timestamp
+from superset.sql.parse import has_aggregate
 from superset.utils import json


@@ -386,13 +387,27 @@ class CreateDatasetMetric(BaseModel):
    """Metric definition for dataset creation."""

    metric_name: str = Field(..., description="Name of the metric")
-    expression: str = Field(..., description="SQL expression for the metric")
+    expression: str = Field(
+        ...,
+        description="Aggregate SQL expression for the metric, e.g. SUM(amount)",
+    )
    verbose_name: str | None = None
    description: str | None = None
    metric_type: str | None = None
    d3format: str | None = None
    warning_text: str | None = None

+    @field_validator("expression")
+    @classmethod
+    def expression_must_aggregate(cls, value: str) -> str:
+        if not has_aggregate(value):
+            raise ValueError(
+                "saved metrics must aggregate rows; wrap a row-level column in "
+                "an aggregate such as MAX(column), or omit the saved metric and "
+                "use the dataset column directly"
+            )
+        return value
+

 class CreateDatasetCalculatedColumn(BaseModel):
    """Calculated column definition for dataset creation."""
@@ -21,6 +21,7 @@ from typing import Any
 from fastmcp import Context
 from superset_core.mcp.decorators import tool, ToolAnnotations

+from superset.exceptions import SupersetGenericDBErrorException
 from superset.extensions import event_logger
 from superset.mcp_service.dataset.schemas import (
    CreateVirtualDatasetRequest,
@@ -67,14 +68,17 @@ def _cleanup_failed_dataset(dataset_id: int) -> None:


 def _update_virtual_dataset(dataset_id: int, update_props: dict[str, Any]) -> Any:
-    from superset.commands.dataset.exceptions import DatasetUpdateFailedError
+    from superset.commands.dataset.exceptions import (
+        DatasetInvalidError,
+        DatasetUpdateFailedError,
+    )
    from superset.commands.dataset.update import UpdateDatasetCommand

    try:
        return UpdateDatasetCommand(dataset_id, update_props).run()
    except Exception as exc:
        _cleanup_failed_dataset(dataset_id)
-        if not isinstance(exc, DatasetUpdateFailedError):
+        if not isinstance(exc, (DatasetInvalidError, DatasetUpdateFailedError)):
            raise DatasetUpdateFailedError() from exc
        raise

@@ -89,7 +93,7 @@ def _update_virtual_dataset(dataset_id: int, update_props: dict[str, Any]) -> An
        destructiveHint=False,
    ),
 )
-async def create_virtual_dataset(
+async def create_virtual_dataset(  # noqa: C901
    request: CreateVirtualDatasetRequest, ctx: Context
 ) -> CreateVirtualDatasetResponse:
    """Save a SQL query as a virtual dataset so it can be charted.
@@ -213,6 +217,18 @@ async def create_virtual_dataset(
            url=None,
            error=f"Failed to update dataset metadata (creation rolled back): {exc}",
        )
+    except SupersetGenericDBErrorException as exc:
+        logger.warning("Virtual dataset SQL validation failed", exc_info=True)
+        await ctx.warning(f"Virtual dataset SQL failed validation: {exc}")
+        return CreateVirtualDatasetResponse(
+            id=None,
+            dataset_name=request.dataset_name,
+            sql=request.sql,
+            database_id=request.database_id,
+            columns=[],
+            url=None,
+            error=f"Dataset SQL could not be executed: {exc}",
+        )
    except Exception as exc:
        await ctx.error(
            f"Unexpected error creating virtual dataset: {type(exc).__name__}: {exc}"
@@ -19,6 +19,7 @@

 from __future__ import annotations

+import ast
 import builtins
 import copy
 import dataclasses
@@ -417,6 +418,52 @@ UUID_NATIVE_TYPE_RE: re.Pattern[str] = re.compile(
 )


+def parse_array_literal(value: Any) -> list[Any]:
+    """
+    Parse a user-entered array literal (e.g. ``['a', 'b']`` or ``[1, 2]``) into a
+    list of elements, for the whole-array (column-level) array operators.
+
+    Accepts either an actual list/tuple, a bracketed literal string (parsed with
+    ``ast.literal_eval``), or a plain scalar (wrapped into a single-element list).
+    Falls back to a single-element list when the string is not a valid literal.
+    """
+    if isinstance(value, (list, tuple)):
+        return list(value)
+    if isinstance(value, str):
+        stripped = value.strip()
+        if stripped.startswith("[") and stripped.endswith("]"):
+            try:
+                parsed = ast.literal_eval(stripped)
+            except (ValueError, SyntaxError):
+                parsed = None
+            if isinstance(parsed, (list, tuple)):
+                return list(parsed)
+    return [value]
+
+
+def coerce_array_values(
+    values: list[Any], element_type: Optional[utils.GenericDataType]
+) -> list[Any]:
+    """
+    Coerce array-element ``values`` to the array column's element type so the
+    emitted literal matches the column. Array columns map to a SQLAlchemy
+    ``String`` type, so values arrive as strings and would otherwise build
+    string literals (e.g. ``array('5')``) that fail against a numeric array on
+    the server. Numeric elements are cast to numbers and boolean elements to
+    booleans; every other element type (string, temporal, enum, unknown) is left
+    untouched.
+
+    :param values: element values entered for an array filter
+    :param element_type: the array's element :class:`GenericDataType`, or None
+    :return: the coerced values
+    """
+    if element_type == utils.GenericDataType.NUMERIC:
+        return [utils.cast_to_num(v) if isinstance(v, str) else v for v in values]
+    if element_type == utils.GenericDataType.BOOLEAN:
+        return [utils.cast_to_boolean(v) if isinstance(v, str) else v for v in values]
+    return values
+
+
 def is_uuid_native_type(native_type: Optional[str]) -> bool:
    """
    Return True if a native column type represents a UUID.
@@ -3652,6 +3699,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
        column_name: str,
        limit: int = 10000,
        denormalize_column: bool = False,
+        array_elements: bool = False,
    ) -> list[Any]:
        # denormalize column name before querying for values
        # unless disabled in the dataset configuration
@@ -3666,13 +3714,25 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
        tp = self.get_template_processor()
        tbl, cte = self.get_from_clause(tp)

+        db_engine_spec = self.database.db_engine_spec
+        value_expr = target_col.get_sqla_col(template_processor=tp)
+        # For element-level operators (Contains any / Contains all) on a
+        # multi-value (array) column, suggest the distinct **elements** rather
+        # than distinct whole arrays by expanding the array first (e.g. ClickHouse
+        # arrayJoin). Only when the engine supports arrays and the column is
+        # actually an array column; otherwise fall back to whole-value suggestions.
+        if array_elements and db_engine_spec.supports_multivalue_columns:
+            col_spec = db_engine_spec.get_column_spec(native_type=target_col.type)
+            if col_spec and col_spec.generic_type == GenericDataType.MULTI_VALUE:
+                value_expr = db_engine_spec.array_explode(value_expr)
+
        qry = (
            sa.select(
                # The alias (label) here is important because some dialects will
                # automatically add a random alias to the projection because of the
                # call to DISTINCT; others will uppercase the column names. This
                # gives us a deterministic column name in the dataframe.
-                target_col.get_sqla_col(template_processor=tp).label("column_values")
+                value_expr.label("column_values")
            )
            .select_from(tbl)
            .distinct()
@@ -4359,7 +4419,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
            elif is_adhoc_column(flt_col):
                try:
                    sqla_col, adhoc_generic_type = self.adhoc_column_to_sqla(
-                        flt_col,
+                        cast("AdhocColumn", flt_col),
                        force_type_check=True,
                        template_processor=template_processor,
                    )
@@ -4433,9 +4493,21 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                    sqla_col = Grouping(sqla_col)
                col_type = col_obj.type if col_obj else None
                col_spec = db_engine_spec.get_column_spec(native_type=col_type)
+                is_multivalue_col = bool(
+                    col_spec and col_spec.generic_type == GenericDataType.MULTI_VALUE
+                )
+                # Element type of an array column (e.g. Array(Int32) -> NUMERIC),
+                # used to coerce filter values before building array expressions.
+                array_element_type = (
+                    db_engine_spec.get_array_element_type(col_type)
+                    if is_multivalue_col
+                    else None
+                )
                is_list_target = op in (
                    utils.FilterOperator.IN,
                    utils.FilterOperator.NOT_IN,
+                    utils.FilterOperator.CONTAINS_ANY,
+                    utils.FilterOperator.CONTAINS_ALL,
                )

                col_advanced_data_type = col_obj.advanced_data_type if col_obj else ""
@@ -4490,7 +4562,56 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                            sqla_col, op, bus_resp["values"]
                        )
                    )
-                elif is_list_target:
+                elif is_multivalue_col and op in {
+                    utils.FilterOperator.EQUALS,
+                    utils.FilterOperator.NOT_EQUALS,
+                    utils.FilterOperator.IN,
+                    utils.FilterOperator.NOT_IN,
+                }:
+                    # Whole-array (column-level) comparison against array
+                    # literal(s). The value is a pasted array literal like
+                    # ``['a', 'b']`` (parsed into elements): ``col = ['a', 'b']``
+                    # for = / !=; for IN / NOT IN each entered value is one such
+                    # array literal (``col IN (['a'], ['b'])``).
+                    if op in {
+                        utils.FilterOperator.EQUALS,
+                        utils.FilterOperator.NOT_EQUALS,
+                    }:
+                        literal = db_engine_spec.array_literal(
+                            coerce_array_values(
+                                parse_array_literal(val), array_element_type
+                            )
+                        )
+                        cond = (
+                            sqla_col != literal
+                            if op == utils.FilterOperator.NOT_EQUALS
+                            else sqla_col == literal
+                        )
+                    else:
+                        candidates: list[Any] = (
+                            list(val) if isinstance(val, (list, tuple)) else [val]
+                        )
+                        cond = sqla_col.in_(
+                            [
+                                db_engine_spec.array_literal(
+                                    coerce_array_values(
+                                        parse_array_literal(candidate),
+                                        array_element_type,
+                                    )
+                                )
+                                for candidate in candidates
+                            ]
+                        )
+                        if op == utils.FilterOperator.NOT_IN:
+                            cond = ~cond
+                    target_clause_list.append(cond)
+                elif op in {
+                    utils.FilterOperator.IN,
+                    utils.FilterOperator.NOT_IN,
+                }:
+                    # CONTAINS_ANY/CONTAINS_ALL also produce a list ``eq`` (they
+                    # are in ``is_list_target``), but are element-level array ops
+                    # handled by their own branch below — not IN.
                    assert isinstance(eq, (tuple, list))
                    if len(eq) == 0:
                        raise QueryObjectValidationError(
@@ -4529,6 +4650,57 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                    target_clause_list.append(
                        db_engine_spec.handle_null_filter(sqla_col, op)
                    )
+                elif op in {
+                    utils.FilterOperator.IS_EMPTY,
+                    utils.FilterOperator.IS_NOT_EMPTY,
+                }:
+                    # Element-level array operators: length(col) == 0 / > 0.
+                    if target_generic_type != GenericDataType.MULTI_VALUE:
+                        raise QueryObjectValidationError(
+                            _(
+                                "The %(op)s operator is only supported for "
+                                "multi-value (array) columns.",
+                                op=op,
+                            )
+                        )
+                    length_expr = db_engine_spec.array_length(sqla_col)
+                    if op == utils.FilterOperator.IS_EMPTY:
+                        target_clause_list.append(length_expr == 0)
+                    else:
+                        target_clause_list.append(length_expr > 0)
+                elif op in {
+                    utils.FilterOperator.LENGTH_EQUALS,
+                    utils.FilterOperator.LENGTH_GREATER_THAN,
+                    utils.FilterOperator.LENGTH_LESS_THAN,
+                    utils.FilterOperator.LENGTH_GREATER_THAN_OR_EQUALS,
+                    utils.FilterOperator.LENGTH_LESS_THAN_OR_EQUALS,
+                }:
+                    # Length filter: compare the array's element count to a
+                    # number, e.g. length(col) > 2.
+                    if target_generic_type != GenericDataType.MULTI_VALUE:
+                        raise QueryObjectValidationError(
+                            _(
+                                "The %(op)s operator is only supported for "
+                                "multi-value (array) columns.",
+                                op=op,
+                            )
+                        )
+                    number = utils.cast_to_num(eq)  # type: ignore[arg-type]
+                    if number is None:
+                        raise QueryObjectValidationError(
+                            _("The Length filter requires a numeric value.")
+                        )
+                    length_expr = db_engine_spec.array_length(sqla_col)
+                    length_comparisons = {
+                        utils.FilterOperator.LENGTH_EQUALS: length_expr == number,
+                        utils.FilterOperator.LENGTH_GREATER_THAN: length_expr > number,
+                        utils.FilterOperator.LENGTH_LESS_THAN: length_expr < number,
+                        utils.FilterOperator.LENGTH_GREATER_THAN_OR_EQUALS: length_expr
+                        >= number,
+                        utils.FilterOperator.LENGTH_LESS_THAN_OR_EQUALS: length_expr
+                        <= number,
+                    }
+                    target_clause_list.append(length_comparisons[op])
                elif op == utils.FilterOperator.IS_TRUE:
                    target_clause_list.append(
                        db_engine_spec.handle_boolean_filter(sqla_col, op, True)
@@ -4586,6 +4758,38 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                            target_clause_list.append(sqla_col.not_like(eq))
                        else:
                            target_clause_list.append(sqla_col.not_ilike(eq))
+                    elif op in {
+                        utils.FilterOperator.CONTAINS_ANY,
+                        utils.FilterOperator.CONTAINS_ALL,
+                    }:
+                        # Element-level array membership. Enforce the target is
+                        # actually a multi-value (array) column (only classified
+                        # MULTI_VALUE on an array-capable engine), guarding against
+                        # payloads that bypass the UI gating.
+                        if target_generic_type != GenericDataType.MULTI_VALUE:
+                            raise QueryObjectValidationError(
+                                _(
+                                    "The %(op)s operator is only supported for "
+                                    "multi-value (array) columns.",
+                                    op=op,
+                                )
+                            )
+                        array_values: list[Any] = coerce_array_values(
+                            list(eq) if isinstance(eq, (list, tuple)) else [eq],
+                            array_element_type,
+                        )
+                        if op == utils.FilterOperator.CONTAINS_ANY:
+                            target_clause_list.append(
+                                db_engine_spec.array_contains_any(
+                                    sqla_col, array_values
+                                )
+                            )
+                        else:
+                            target_clause_list.append(
+                                db_engine_spec.array_contains_all(
+                                    sqla_col, array_values
+                                )
+                            )
                    elif (
                        op == utils.FilterOperator.TEMPORAL_RANGE
                        and isinstance(eq, str)
@@ -209,7 +209,7 @@ class GenericDataType(IntEnum):
    STRING = 1
    TEMPORAL = 2
    BOOLEAN = 3
-    # ARRAY = 4     # Mapping all the complex data types to STRING for now
+    MULTI_VALUE = 4  # array-typed columns (e.g. ClickHouse Array, Postgres ARRAY)
    # JSON = 5      # and leaving these as a reminder.
    # MAP = 6
    # ROW = 7
@@ -299,6 +299,17 @@ class FilterOperator(StrEnum):
    IS_TRUE = "IS TRUE"
    IS_FALSE = "IS FALSE"
    TEMPORAL_RANGE = "TEMPORAL_RANGE"
+    # Element-level operators for MULTI_VALUE (array) columns
+    CONTAINS_ANY = "CONTAINS_ANY"
+    CONTAINS_ALL = "CONTAINS_ALL"
+    IS_EMPTY = "IS_EMPTY"
+    IS_NOT_EMPTY = "IS_NOT_EMPTY"
+    # Length (element-count) comparison operators for array columns
+    LENGTH_EQUALS = "LENGTH_EQUALS"
+    LENGTH_GREATER_THAN = "LENGTH_GREATER_THAN"
+    LENGTH_LESS_THAN = "LENGTH_LESS_THAN"
+    LENGTH_GREATER_THAN_OR_EQUALS = "LENGTH_GREATER_THAN_OR_EQUALS"
+    LENGTH_LESS_THAN_OR_EQUALS = "LENGTH_LESS_THAN_OR_EQUALS"


 class FilterStringOperators(StrEnum):
@@ -317,6 +328,15 @@ class FilterStringOperators(StrEnum):
    LATEST_PARTITION = ("LATEST_PARTITION",)
    IS_TRUE = ("IS_TRUE",)
    IS_FALSE = ("IS_FALSE",)
+    CONTAINS_ANY = ("CONTAINS_ANY",)
+    CONTAINS_ALL = ("CONTAINS_ALL",)
+    IS_EMPTY = ("IS_EMPTY",)
+    IS_NOT_EMPTY = ("IS_NOT_EMPTY",)
+    LENGTH_EQUALS = ("LENGTH_EQUALS",)
+    LENGTH_GREATER_THAN = ("LENGTH_GREATER_THAN",)
+    LENGTH_LESS_THAN = ("LENGTH_LESS_THAN",)
+    LENGTH_GREATER_THAN_OR_EQUALS = ("LENGTH_GREATER_THAN_OR_EQUALS",)
+    LENGTH_LESS_THAN_OR_EQUALS = ("LENGTH_LESS_THAN_OR_EQUALS",)


 class PostProcessingBoxplotWhiskerType(StrEnum):
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from collections.abc import Sequence
-from functools import partial
+from functools import partial, wraps
 from typing import Any, Callable

 import numpy as np
@@ -122,6 +122,10 @@ def scalar_to_sequence(val: Any) -> Sequence[str]:

 def validate_column_args(*argnames: str) -> Callable[..., Any]:
    def wrapper(func: Callable[..., Any]) -> Callable[..., Any]:
+        # `wraps` keeps `func` reachable through `__wrapped__`, so that
+        # `inspect.signature` reports the parameters of the decorated operation
+        # rather than the `(df, **options)` of this wrapper.
+        @wraps(func)
        def wrapped(df: DataFrame, **options: Any) -> Any:
            if _is_multi_index_on_columns(df):
                # MultiIndex column validate first level
@@ -56,6 +56,7 @@ from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,  # noqa: F401
    load_birth_names_data,  # noqa: F401
 )
+from tests.integration_tests.fixtures.client import client  # noqa: F401
 from tests.integration_tests.fixtures.energy_dashboard import (
    load_energy_table_data,  # noqa: F401
    load_energy_table_with_slice,  # noqa: F401
@@ -65,6 +66,10 @@ from tests.integration_tests.fixtures.importexport import (
    database_config,
    dataset_config,
 )
+from tests.integration_tests.fixtures.lineage import (
+    inject_expected_chart_lineage,  # noqa: F401
+    lineage_test_data,  # noqa: F401
+)
 from tests.integration_tests.fixtures.tags import (
    create_custom_tags,  # noqa: F401
    get_filter_params,
@@ -2665,3 +2670,30 @@ class TestChartApi(ApiEditorsTestCaseMixin, InsertChartMixin, SupersetTestCase):
        self.login(ADMIN_USERNAME)
        rv = self.client.get("api/v1/chart/related/editors")
        assert rv.status_code == 200
+
+    @pytest.mark.usefixtures("inject_expected_chart_lineage")
+    def test_get_chart_lineage(self):
+        """
+        Chart API: Test get chart lineage
+        """
+        self.login(ADMIN_USERNAME)
+        chart_id = self.chart_lineage["chart_id"]
+        expected = self.chart_lineage["expected"]
+
+        uri = f"api/v1/chart/{chart_id}/lineage"
+        rv = self.get_assert_metric(uri, "lineage")
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+
+        # The lineage payload is wrapped under "result"
+        assert data["result"] == expected
+
+    def test_get_chart_lineage_not_found(self):
+        """
+        Chart API: Test get chart lineage with non-existent chart
+        """
+        self.login(ADMIN_USERNAME)
+        uri = "api/v1/chart/99999/lineage"
+        rv = self.client.get(uri)
+        assert rv.status_code == 404
@@ -74,6 +74,11 @@ from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,  # noqa: F401
    load_birth_names_data,  # noqa: F401
 )
+from tests.integration_tests.fixtures.client import client  # noqa: F401
+from tests.integration_tests.fixtures.lineage import (
+    inject_expected_dashboard_lineage,  # noqa: F401
+    lineage_test_data,  # noqa: F401
+)
 from tests.integration_tests.fixtures.world_bank_dashboard import (
    load_world_bank_dashboard_with_slices,  # noqa: F401
    load_world_bank_data,  # noqa: F401
@@ -4473,6 +4478,33 @@ class TestDashboardApi(ApiEditorsTestCaseMixin, InsertChartMixin, SupersetTestCa
            db.session.delete(dashboard)
            db.session.commit()

+    @pytest.mark.usefixtures("inject_expected_dashboard_lineage")
+    def test_get_dashboard_lineage(self):
+        """
+        Dashboard API: Test get dashboard lineage
+        """
+        self.login(ADMIN_USERNAME)
+        dashboard_id = self.dashboard_lineage["dashboard_id"]
+        expected = self.dashboard_lineage["expected"]
+
+        uri = f"api/v1/dashboard/{dashboard_id}/lineage"
+        rv = self.get_assert_metric(uri, "lineage")
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+
+        # The lineage payload is wrapped under "result"
+        assert data["result"] == expected
+
+    def test_get_dashboard_lineage_not_found(self):
+        """
+        Dashboard API: Test get dashboard lineage with non-existent dashboard
+        """
+        self.login(ADMIN_USERNAME)
+        uri = "api/v1/dashboard/99999/lineage"
+        rv = self.client.get(uri)
+        assert rv.status_code == 404
+

 class TestDashboardCustomTagsFiltering(SupersetTestCase):
    """Test dashboard list API tags field behavior.
@@ -62,6 +62,7 @@ from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,  # noqa: F401
    load_birth_names_data,  # noqa: F401
 )
+from tests.integration_tests.fixtures.client import client  # noqa: F401
 from tests.integration_tests.fixtures.energy_dashboard import (
    load_energy_table_data,  # noqa: F401
    load_energy_table_with_slice,  # noqa: F401
@@ -71,6 +72,10 @@ from tests.integration_tests.fixtures.importexport import (
    dataset_config,
    dataset_ui_export,
 )
+from tests.integration_tests.fixtures.lineage import (
+    inject_expected_dataset_lineage,  # noqa: F401
+    lineage_test_data,  # noqa: F401
+)


 class TestDatasetApi(SupersetTestCase):
@@ -3634,3 +3639,215 @@ class TestDatasetApi(SupersetTestCase):
            assert rv.status_code == 403

        self.items_to_delete = [dash, chart, dataset, dashboard_dataset]
+
+    @with_feature_flags(DASHBOARD_RBAC=True)
+    def test_get_drill_info_dashboard_rbac_access_granted(self):
+        """
+        Dataset API: Test drill_info with dashboard parameter when user has access
+        via the DASHBOARD_RBAC FF.
+        """
+        with self.temporary_user(
+            clone_user=security_manager.find_user(username=GAMMA_USERNAME)
+        ) as test_user:
+            user_role_ids = [role.id for role in test_user.roles]
+            # Login as admin to avoid FK issues during temp account deletion
+            self.login(ADMIN_USERNAME)
+            dataset = self.insert_dataset(
+                table_name="test_rbac_dataset",
+                owners=[],
+                columns=[
+                    TableColumn(
+                        column_name="restricted_column",
+                        type="VARCHAR(255)",
+                        verbose_name="Restricted Column",
+                        groupby=True,
+                    ),
+                ],
+                fetch_metadata=False,
+            )
+            chart = self.insert_chart("Test RBAC Chart", dataset.id)
+            dash = self.insert_dashboard(
+                "RBAC Test Dashboard",
+                "rbac-test-dashboard",
+                [],
+                roles=user_role_ids,
+                slices=[chart],
+                published=True,
+            )
+
+            self.logout()
+            self.login(test_user.username)
+
+            uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
+            rv = self.client.get(uri)
+
+            assert rv.status_code == 200
+            data = json.loads(rv.data.decode("utf-8"))
+            result = data["result"]
+
+            assert "created_by" in result
+            assert "created_on_humanized" in result
+            assert "changed_by" in result
+            assert "changed_on_humanized" in result
+            assert result["id"] == dataset.id
+            assert result["table_name"] == "test_rbac_dataset"
+            assert len(result["columns"]) == 1
+            assert result["columns"][0]["column_name"] == "restricted_column"
+
+        self.items_to_delete = [dash, chart, dataset]
+
+    @with_feature_flags(DASHBOARD_RBAC=True)
+    def test_get_drill_info_dashboard_rbac_no_perm_to_drill(self):
+        """
+        Dataset API: Test drill_info with dashboard parameter when user has
+        no permission to access the API.
+        """
+        with self.temporary_user(
+            clone_user=security_manager.find_user(username=GAMMA_USERNAME),
+            pvms_to_remove=[("can_get_drill_info", "Dataset")],
+        ) as test_user:
+            user_role_ids = [role.id for role in test_user.roles]
+            self.login(ADMIN_USERNAME)
+
+            dataset = self.insert_dataset(
+                table_name="test_rbac_dataset_denied",
+                owners=[],
+                columns=[
+                    TableColumn(
+                        column_name="restricted_column",
+                        type="VARCHAR(255)",
+                        groupby=True,
+                    ),
+                ],
+                fetch_metadata=False,
+            )
+            chart = self.insert_chart("Test RBAC Chart second", dataset.id)
+            dash = self.insert_dashboard(
+                "RBAC Test Dashboard 2",
+                "rbac-test-dashboard-2",
+                [],
+                slices=[chart],
+                roles=user_role_ids,
+                published=True,
+            )
+
+            self.logout()
+            self.login(test_user.username)
+
+            uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
+            rv = self.client.get(uri)
+
+            assert rv.status_code == 403
+
+        self.items_to_delete = [dash, chart, dataset]
+
+    @with_feature_flags(DASHBOARD_RBAC=True)
+    def test_get_drill_info_dashboard_rbac_no_access_on_dashboard(self):
+        """
+        Dataset API: Test drill_info with dashboard parameter when user has
+        no access to the dashboard.
+        """
+        dataset = self.insert_dataset(
+            table_name="test_rbac_dataset_denied",
+            owners=[],
+            columns=[
+                TableColumn(
+                    column_name="restricted_column",
+                    type="VARCHAR(255)",
+                    groupby=True,
+                ),
+            ],
+            fetch_metadata=False,
+        )
+        chart = self.insert_chart("Test RBAC Chart second", dataset.id)
+        dash = self.insert_dashboard(
+            "RBAC Test Dashboard 2",
+            "rbac-test-dashboard-2",
+            [],
+            slices=[chart],
+            roles=[],
+            published=True,
+        )
+
+        with self.temporary_user(
+            clone_user=security_manager.find_user(username=GAMMA_USERNAME),
+            login=True,
+            username="test_new_account",
+        ):
+            uri = f"api/v1/dataset/{dataset.id}/drill_info/?q=(dashboard_id:{dash.id})"
+            rv = self.client.get(uri)
+
+            assert rv.status_code == 403
+
+        self.items_to_delete = [dash, chart, dataset]
+
+    @with_feature_flags(DASHBOARD_RBAC=True)
+    def test_get_drill_info_dashboard_rbac_no_dashboard_id(self):
+        """
+        Dataset API: Test drill_info without dashboard ID parameter falls back
+        to regular access control.
+        """
+        with self.temporary_user(
+            clone_user=security_manager.find_user(username=GAMMA_USERNAME),
+        ) as test_user:
+            self.login(ADMIN_USERNAME)
+            user_role_ids = [role.id for role in test_user.roles]
+
+            dataset = self.insert_dataset(
+                table_name="test_no_dashboard_id",
+                owners=[],
+                columns=[
+                    TableColumn(
+                        column_name="restricted_column",
+                        type="VARCHAR(255)",
+                        groupby=True,
+                    ),
+                ],
+                fetch_metadata=False,
+            )
+            chart = self.insert_chart("Test RBAC Chart second", dataset.id)
+            dashboard = self.insert_dashboard(
+                "RBAC Test Dashboard 2",
+                "rbac-test-dashboard-2",
+                [],
+                slices=[chart],
+                roles=user_role_ids,
+                published=True,
+            )
+
+            self.logout()
+            self.login(test_user.username)
+
+            uri = f"api/v1/dataset/{dataset.id}/drill_info/"
+            rv = self.client.get(uri)
+
+            assert rv.status_code == 404
+
+        self.items_to_delete = [dashboard, chart, dataset]
+
+    @pytest.mark.usefixtures("inject_expected_dataset_lineage")
+    def test_get_dataset_lineage(self):
+        """
+        Dataset API: Test get dataset lineage
+        """
+        self.login(ADMIN_USERNAME)
+        dataset_id = self.dataset_lineage["dataset_id"]
+        expected = self.dataset_lineage["expected"]
+
+        uri = f"api/v1/dataset/{dataset_id}/lineage"
+        rv = self.get_assert_metric(uri, "lineage")
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+
+        # The lineage payload is wrapped under "result"
+        assert data["result"] == expected
+
+    def test_get_dataset_lineage_not_found(self):
+        """
+        Dataset API: Test get dataset lineage with non-existent dataset
+        """
+        self.login(ADMIN_USERNAME)
+        uri = "api/v1/dataset/99999/lineage"
+        rv = self.client.get(uri)
+        assert rv.status_code == 404
@@ -154,8 +154,22 @@ class TestDatasourceApi(SupersetTestCase):
            column_name="col2",
            limit=10000,
            denormalize_column=False,
+            array_elements=False,
        )

+    @pytest.mark.usefixtures("app_context", "virtual_dataset")
+    @patch("superset.models.helpers.ExploreMixin.values_for_column")
+    def test_get_column_values_array_elements_param(self, values_for_column_mock):
+        # The ?array_elements=true param (Contains any/all) is threaded through
+        # so array columns can suggest individual elements.
+        self.login(ADMIN_USERNAME)
+        table = self.get_virtual_dataset()
+        self.client.get(
+            f"api/v1/datasource/table/{table.id}/column/col2/values/"
+            "?array_elements=true"
+        )
+        assert values_for_column_mock.call_args.kwargs["array_elements"] is True
+
    @pytest.mark.usefixtures("app_context", "virtual_dataset")
    @patch("superset.db_engine_specs.base.BaseEngineSpec.denormalize_name")
    def test_get_column_values_not_denormalize_column(self, denormalize_name_mock):
@@ -176,6 +190,7 @@ class TestDatasourceApi(SupersetTestCase):
            column_name="col2",
            limit=10000,
            denormalize_column=True,
+            array_elements=False,
        )

    @pytest.mark.usefixtures("app_context", "virtual_dataset")
@@ -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.
+import pytest
+
+from superset import db
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+from superset.utils.database import get_example_database
+from tests.integration_tests.dashboard_utils import create_table_metadata
+
+
+@pytest.fixture
+def lineage_test_data(app_context, load_birth_names_data):
+    """
+    Base fixture that creates a simple lineage structure and returns
+    the created entities (database, dataset, charts, dashboard).
+    """
+    database = get_example_database()
+
+    # Create dataset
+    dataset = create_table_metadata(
+        table_name="lineage_test_dataset",
+        database=database,
+    )
+    db.session.add(dataset)
+    db.session.flush()
+
+    # Create charts
+    chart1 = Slice(
+        slice_name="Lineage Test Chart 1",
+        viz_type="table",
+        datasource_id=dataset.id,
+        datasource_type="table",
+        params="{}",
+    )
+    chart2 = Slice(
+        slice_name="Lineage Test Chart 2",
+        viz_type="pie",
+        datasource_id=dataset.id,
+        datasource_type="table",
+        params="{}",
+    )
+    db.session.add(chart1)
+    db.session.add(chart2)
+    db.session.flush()
+
+    # Create dashboard with charts
+    dashboard = Dashboard(
+        dashboard_title="Lineage Test Dashboard",
+        slug="lineage-test-dashboard",
+        slices=[chart1, chart2],
+        published=True,
+    )
+    db.session.add(dashboard)
+    db.session.commit()
+
+    # Return the created entities
+    result = {
+        "database": database,
+        "dataset": dataset,
+        "charts": [chart1, chart2],
+        "dashboard": dashboard,
+    }
+
+    yield result
+
+    # Cleanup
+    db.session.delete(dashboard)
+    db.session.delete(chart1)
+    db.session.delete(chart2)
+    for col in dataset.columns + dataset.metrics:
+        db.session.delete(col)
+    db.session.delete(dataset)
+    db.session.commit()
+
+
+@pytest.fixture(autouse=False)
+def inject_expected_dataset_lineage(request, lineage_test_data):
+    """
+    Injects dataset lineage data into test class instance.
+    """
+    dataset = lineage_test_data["dataset"]
+    database = lineage_test_data["database"]
+    charts = lineage_test_data["charts"]
+    dashboard = lineage_test_data["dashboard"]
+
+    request.instance.dataset_lineage = {
+        "dataset_id": dataset.id,
+        "expected": {
+            "dataset": {
+                "id": dataset.id,
+                "name": dataset.name,
+                "schema": dataset.schema,
+                "table_name": dataset.table_name,
+                "database_id": database.id,
+                "database_name": database.database_name,
+            },
+            "upstream": {
+                "database": {
+                    "id": database.id,
+                    "database_name": database.database_name,
+                    "backend": database.backend,
+                }
+            },
+            "downstream": {
+                "charts": {
+                    "count": 2,
+                    "result": [
+                        {
+                            "id": charts[0].id,
+                            "slice_name": charts[0].slice_name,
+                            "viz_type": charts[0].viz_type,
+                            "dashboard_ids": [dashboard.id],
+                        },
+                        {
+                            "id": charts[1].id,
+                            "slice_name": charts[1].slice_name,
+                            "viz_type": charts[1].viz_type,
+                            "dashboard_ids": [dashboard.id],
+                        },
+                    ],
+                },
+                "dashboards": {
+                    "count": 1,
+                    "result": [
+                        {
+                            "id": dashboard.id,
+                            "title": dashboard.dashboard_title,
+                            "slug": dashboard.slug,
+                            "chart_ids": sorted([charts[0].id, charts[1].id]),
+                        }
+                    ],
+                },
+            },
+        },
+    }
+
+
+@pytest.fixture(autouse=False)
+def inject_expected_chart_lineage(request, lineage_test_data):
+    """
+    Injects chart lineage data into test class instance.
+    """
+    dataset = lineage_test_data["dataset"]
+    database = lineage_test_data["database"]
+    chart = lineage_test_data["charts"][0]  # Use first chart
+    dashboard = lineage_test_data["dashboard"]
+
+    request.instance.chart_lineage = {
+        "chart_id": chart.id,
+        "expected": {
+            "chart": {
+                "id": chart.id,
+                "slice_name": chart.slice_name,
+                "viz_type": chart.viz_type,
+            },
+            "upstream": {
+                "dataset": {
+                    "id": dataset.id,
+                    "name": dataset.name,
+                    "schema": dataset.schema,
+                    "table_name": dataset.table_name,
+                    "database_id": database.id,
+                    "database_name": database.database_name,
+                },
+                "database": {
+                    "id": database.id,
+                    "database_name": database.database_name,
+                    "backend": database.backend,
+                },
+            },
+            "downstream": {
+                "dashboards": {
+                    "count": 1,
+                    "result": [
+                        {
+                            "id": dashboard.id,
+                            "title": dashboard.dashboard_title,
+                            "slug": dashboard.slug,
+                        }
+                    ],
+                }
+            },
+        },
+    }
+
+
+@pytest.fixture(autouse=False)
+def inject_expected_dashboard_lineage(request, lineage_test_data):
+    """
+    Injects dashboard lineage data into test class instance.
+    """
+    dataset = lineage_test_data["dataset"]
+    database = lineage_test_data["database"]
+    charts = lineage_test_data["charts"]
+    dashboard = lineage_test_data["dashboard"]
+
+    request.instance.dashboard_lineage = {
+        "dashboard_id": dashboard.id,
+        "expected": {
+            "dashboard": {
+                "id": dashboard.id,
+                "title": dashboard.dashboard_title,
+                "slug": dashboard.slug,
+                "published": dashboard.published,
+            },
+            "upstream": {
+                "charts": {
+                    "count": 2,
+                    "result": [
+                        {
+                            "id": charts[0].id,
+                            "slice_name": charts[0].slice_name,
+                            "viz_type": charts[0].viz_type,
+                            "dataset_id": dataset.id,
+                        },
+                        {
+                            "id": charts[1].id,
+                            "slice_name": charts[1].slice_name,
+                            "viz_type": charts[1].viz_type,
+                            "dataset_id": dataset.id,
+                        },
+                    ],
+                },
+                "datasets": {
+                    "count": 1,
+                    "result": [
+                        {
+                            "id": dataset.id,
+                            "name": dataset.name,
+                            "schema": dataset.schema,
+                            "table_name": dataset.table_name,
+                            "database_id": database.id,
+                            "database_name": database.database_name,
+                            "chart_ids": sorted([charts[0].id, charts[1].id]),
+                        }
+                    ],
+                },
+                "databases": {
+                    "count": 1,
+                    "result": [
+                        {
+                            "id": database.id,
+                            "database_name": database.database_name,
+                            "backend": database.backend,
+                        }
+                    ],
+                },
+            },
+            "downstream": None,
+        },
+    }
@@ -1308,3 +1308,152 @@ def test_column_ordering_without_chart_flag(login_as_admin):
    finally:
        db.session.delete(table)
        db.session.commit()
+
+
+def _multivalue_table() -> SqlaTable:
+    """A dataset with an ``Array(String)`` column, for multi-value query tests.
+
+    Built over the example database but never executed — the tests only compile
+    the generated SQL, so the backing table need not physically exist.
+    """
+    columns = [
+        TableColumn(column_name="skills", type="Array(String)"),
+        TableColumn(column_name="city", type="VARCHAR(255)"),
+    ]
+    return SqlaTable(
+        table_name="test_multivalue_jobs",
+        database=get_example_database(),
+        columns=columns,
+        metrics=[SqlMetric(metric_name="count", expression="COUNT(*)")],
+    )
+
+
+def _multivalue_query(
+    *,
+    filters: list[dict[str, Any]] | None = None,
+    groupby: list[Any] | None = None,
+) -> dict[str, Any]:
+    return {
+        "granularity": None,
+        "from_dttm": None,
+        "to_dttm": None,
+        "is_timeseries": False,
+        "groupby": groupby if groupby is not None else ["city"],
+        "metrics": ["count"],
+        "filter": filters or [],
+        "extras": {},
+    }
+
+
+def _compile(table: SqlaTable, query_obj: dict[str, Any]) -> str:
+    from superset.db_engine_specs.clickhouse import ClickHouseEngineSpec
+
+    with patch.object(
+        SqlaTable, "db_engine_spec", property(lambda self: ClickHouseEngineSpec)
+    ):
+        sqla_query = table.get_sqla_query(**query_obj)
+        return table.database.compile_sqla_query(sqla_query.sqla_query).lower()
+
+
+@pytest.mark.usefixtures("app_context")
+def test_multivalue_contains_any_generates_native_sql():
+    """CONTAINS_ANY compiles to ``hasAny(col, array(...))``."""
+    table = _multivalue_table()
+    sql = _compile(
+        table,
+        _multivalue_query(
+            filters=[
+                {
+                    "col": "skills",
+                    "op": FilterOperator.CONTAINS_ANY.value,
+                    "val": ["Driver", "Cook"],
+                }
+            ]
+        ),
+    )
+    assert "hasany(skills" in sql
+
+
+@pytest.mark.usefixtures("app_context")
+def test_multivalue_contains_all_generates_native_sql():
+    """CONTAINS_ALL compiles to ``hasAll(col, array(...))``."""
+    table = _multivalue_table()
+    sql = _compile(
+        table,
+        _multivalue_query(
+            filters=[
+                {
+                    "col": "skills",
+                    "op": FilterOperator.CONTAINS_ALL.value,
+                    "val": ["Driver", "Cook"],
+                }
+            ]
+        ),
+    )
+    assert "hasall(skills" in sql
+
+
+@pytest.mark.usefixtures("app_context")
+def test_multivalue_is_empty_generates_native_sql():
+    """IS_EMPTY compiles to ``length(col) = 0``."""
+    table = _multivalue_table()
+    sql = _compile(
+        table,
+        _multivalue_query(
+            filters=[{"col": "skills", "op": FilterOperator.IS_EMPTY.value}]
+        ),
+    )
+    assert "length(skills) = 0" in sql
+
+
+@pytest.mark.usefixtures("app_context")
+def test_multivalue_length_filter_generates_native_sql():
+    """A LENGTH_GREATER_THAN filter compiles to ``length(col) > N``."""
+    table = _multivalue_table()
+    sql = _compile(
+        table,
+        _multivalue_query(
+            filters=[
+                {
+                    "col": "skills",
+                    "op": FilterOperator.LENGTH_GREATER_THAN.value,
+                    "val": 2,
+                }
+            ]
+        ),
+    )
+    assert "length(skills) > 2" in sql
+
+
+@pytest.mark.usefixtures("app_context")
+def test_multivalue_contains_unsupported_engine_raises():
+    """CONTAINS_ANY on an engine without array support is rejected."""
+    table = _multivalue_table()
+    query_obj = _multivalue_query(
+        filters=[
+            {
+                "col": "skills",
+                "op": FilterOperator.CONTAINS_ANY.value,
+                "val": ["Driver"],
+            }
+        ]
+    )
+    with pytest.raises(QueryObjectValidationError):
+        table.get_sqla_query(**query_obj)
+
+
+@pytest.mark.usefixtures("app_context")
+def test_multivalue_length_filter_unsupported_engine_raises():
+    """A Length filter on an engine without array support is rejected."""
+    table = _multivalue_table()
+    query_obj = _multivalue_query(
+        filters=[
+            {
+                "col": "skills",
+                "op": FilterOperator.LENGTH_GREATER_THAN.value,
+                "val": 2,
+            }
+        ]
+    )
+    with pytest.raises(QueryObjectValidationError):
+        table.get_sqla_query(**query_obj)
@@ -1490,3 +1490,21 @@ def test_get_public_information_exposes_ansi_identifier_quote() -> None:
        "end": '"',
        "escape_by_doubling": True,
    }
+
+
+def test_multivalue_columns_disabled_by_default() -> None:
+    """Engines must opt in to multi-value support; base defaults to off."""
+    assert BaseEngineSpec.supports_multivalue_columns is False
+
+
+@pytest.mark.parametrize(
+    "method", ["array_contains_any", "array_contains_all", "array_length"]
+)
+def test_array_capabilities_raise_when_unsupported(method: str) -> None:
+    """Array capability methods raise NotImplementedError unless overridden."""
+    from sqlalchemy import column
+
+    fn = getattr(BaseEngineSpec, method)
+    args = (column("c"), ["v"]) if "contains" in method else (column("c"),)
+    with pytest.raises(NotImplementedError):
+        fn(*args)
@@ -143,7 +143,30 @@ def test_connect_convert_dttm(
            GenericDataType.STRING,
            False,
        ),
-        ("Array(UInt8)", String, None, GenericDataType.STRING, False),
+        ("Array(UInt8)", String, None, GenericDataType.MULTI_VALUE, False),
+        ("Array(String)", String, None, GenericDataType.MULTI_VALUE, False),
+        ("Array(UInt64)", String, None, GenericDataType.MULTI_VALUE, False),
+        (
+            "Array(LowCardinality(String))",
+            String,
+            None,
+            GenericDataType.MULTI_VALUE,
+            False,
+        ),
+        # Array(Enum(...)) is a real array and must classify as MULTI_VALUE, not
+        # get short-circuited by the Enum rule (the anchored ^Array\( pattern is
+        # ordered before the Enum entry).
+        (
+            "Array(Enum8('a' = 1, 'b' = 2))",
+            String,
+            None,
+            GenericDataType.MULTI_VALUE,
+            False,
+        ),
+        # Arrays nested inside Map/Tuple are not top-level array columns; the
+        # anchored pattern must not over-match them into MULTI_VALUE.
+        ("Map(String, Array(String))", String, None, GenericDataType.STRING, False),
+        ("Tuple(Array(String))", String, None, GenericDataType.STRING, False),
        ("Enum('hello', 'world')", String, None, GenericDataType.STRING, False),
        ("Enum('UInt32', 'Bool')", String, None, GenericDataType.STRING, False),
        (
@@ -630,3 +653,115 @@ def test_use_equality_for_boolean_filters_property() -> None:
    from superset.db_engine_specs.clickhouse import ClickHouseBaseEngineSpec

    assert ClickHouseBaseEngineSpec.use_equality_for_boolean_filters is True
+
+
+def _compile(expr) -> str:
+    return str(expr.compile(compile_kwargs={"literal_binds": True}))
+
+
+def test_clickhouse_supports_multivalue_columns() -> None:
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    assert spec.supports_multivalue_columns is True
+
+
+def test_multivalue_contains_any_sql() -> None:
+    from sqlalchemy import column
+
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    expr = spec.array_contains_any(column("skills"), ["Driver", "Cook"])
+    assert _compile(expr) == "hasAny(skills, array('Driver', 'Cook'))"
+
+
+def test_multivalue_contains_all_sql() -> None:
+    from sqlalchemy import column
+
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    expr = spec.array_contains_all(column("skills"), ["Driver", "Cook"])
+    assert _compile(expr) == "hasAll(skills, array('Driver', 'Cook'))"
+
+
+def test_multivalue_contains_binds_parameters() -> None:
+    """Values must be bound parameters, not inlined (SQL-injection safety)."""
+    from sqlalchemy import column
+
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    expr = spec.array_contains_any(column("skills"), ["Driver"])
+    compiled = expr.compile()
+    assert "Driver" not in str(compiled)
+    assert "Driver" in compiled.params.values()
+
+
+def test_multivalue_length_sql() -> None:
+    from sqlalchemy import column
+
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    expr = spec.array_length(column("skills"))
+    assert _compile(expr) == "length(skills)"
+
+
+@pytest.mark.parametrize(
+    "native_type,expected",
+    [
+        ("Array(String)", GenericDataType.STRING),
+        ("Array(Int32)", GenericDataType.NUMERIC),
+        ("Array(UInt64)", GenericDataType.NUMERIC),
+        ("Array(Decimal(10, 2))", GenericDataType.NUMERIC),
+        ("Array(DateTime)", GenericDataType.TEMPORAL),
+        ("Array(Enum8('a' = 1))", GenericDataType.STRING),
+        # Wrappers around the element type don't change the generic type.
+        ("Array(Nullable(Int64))", GenericDataType.NUMERIC),
+        ("Array(LowCardinality(String))", GenericDataType.STRING),
+        # Non-array / nested-array types have no array element type.
+        ("String", None),
+        ("Map(String, Array(String))", None),
+    ],
+)
+def test_multivalue_get_array_element_type(
+    native_type: str, expected: GenericDataType | None
+) -> None:
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    assert spec.get_array_element_type(native_type) == expected
+
+
+def test_multivalue_array_explode_sql() -> None:
+    """array_explode compiles to ``arrayJoin(col)`` (element expansion)."""
+    from sqlalchemy import column
+
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    expr = spec.array_explode(column("scores"))
+    assert _compile(expr) == "arrayJoin(scores)"
+
+
+def test_multivalue_contains_any_numeric_coercion_sql() -> None:
+    """Numeric-array element values must render as numbers, not quoted strings."""
+    from sqlalchemy import column
+
+    from superset.db_engine_specs.clickhouse import (  # noqa: N813
+        ClickHouseEngineSpec as spec,
+    )
+
+    # Simulate values already coerced to numbers (as helpers.py does via the
+    # element type) and confirm the emitted array literal is numeric.
+    expr = spec.array_contains_any(column("scores"), [5, 6])
+    assert _compile(expr) == "hasAny(scores, array(5, 6))"
@@ -1981,6 +1981,23 @@ def test_create_virtual_dataset_request_optional_fields() -> None:
    assert req.description == "A virtual dataset"


+def test_create_virtual_dataset_rejects_non_aggregate_saved_metric() -> None:
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError, match="saved metrics must aggregate rows"):
+        CreateVirtualDatasetRequest(
+            database_id=1,
+            sql="SELECT needed_operators FROM staffing",
+            dataset_name="Staffing",
+            metrics=[
+                {
+                    "metric_name": "needed_operators",
+                    "expression": "needed_operators",
+                }
+            ],
+        )
+
+
 # --- Tool logic tests ---


@@ -2119,6 +2136,39 @@ async def test_create_virtual_dataset_create_failed(mcp_server: object) -> None:
    assert "Failed to create dataset" in data["error"]


+@pytest.mark.asyncio
+async def test_create_virtual_dataset_sql_error_is_actionable(
+    mcp_server: object,
+) -> None:
+    """Warehouse SQL errors are recoverable tool results, not adapter crashes."""
+    from superset.exceptions import SupersetGenericDBErrorException
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = SupersetGenericDBErrorException(
+        "Invalid column name 'missing_value'"
+    )
+
+    with patch(
+        "superset.commands.dataset.create.CreateDatasetCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = CreateVirtualDatasetRequest(
+                database_id=1,
+                sql="SELECT missing_value FROM sample_events",
+                dataset_name="Test",
+            )
+            result = await client.call_tool(
+                "create_virtual_dataset", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert data["columns"] == []
+    assert data["error"] is not None
+    assert "Invalid column name" in data["error"]
+
+
@pytest.mark.asyncio
 async def test_create_virtual_dataset_permission_denied(mcp_server: object) -> None:
    """SQL access denied surfaces as DatasetInvalidError with id=None."""
@@ -2289,7 +2339,13 @@ async def test_create_virtual_dataset_update_failure_rollback(
    if exception_to_raise == "DatasetUpdateFailedError":
        mock_update_instance.run.side_effect = DatasetUpdateFailedError()
    else:
-        mock_update_instance.run.side_effect = DatasetInvalidError()
+        from superset.commands.dataset.exceptions import (
+            DatasetColumnsExistsValidationError,
+        )
+
+        invalid_error = DatasetInvalidError()
+        invalid_error.append(DatasetColumnsExistsValidationError())
+        mock_update_instance.run.side_effect = invalid_error
    mock_update_cls = MagicMock(return_value=mock_update_instance)

    mock_delete_instance = MagicMock()
@@ -2336,7 +2392,11 @@ async def test_create_virtual_dataset_update_failure_rollback(
    # Verify the error response
    data = json.loads(result.content[0].text)
    assert data["id"] is None
-    assert "creation rolled back" in data["error"]
+    if exception_to_raise == "DatasetInvalidError":
+        assert "columns" in data["error"]
+        assert "already exist" in data["error"]
+    else:
+        assert "creation rolled back" in data["error"]


@pytest.mark.asyncio
@@ -0,0 +1,274 @@
+# 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.
+"""Element-level array filter operators (Contains Any/All, Is empty/not empty)."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+import pytest
+from flask import Flask
+from pytest_mock import MockerFixture
+
+from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn
+from superset.exceptions import QueryObjectValidationError
+from superset.models.core import Database
+from superset.superset_typing import QueryObjectDict
+from superset.utils.core import FilterOperator
+
+
+def _make_dataset(mocker: MockerFixture) -> SqlaTable:
+    database = Database(id=1, database_name="test_db", sqlalchemy_uri="sqlite://")
+    columns = [
+        TableColumn(column_name="skills", type="Array(String)"),
+        TableColumn(column_name="scores", type="Array(Int32)"),
+        TableColumn(column_name="city", type="VARCHAR(100)"),
+    ]
+    dataset = SqlaTable(
+        table_name="jobs",
+        columns=columns,
+        database=database,
+        metrics=[SqlMetric(metric_name="count", expression="COUNT(*)")],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
+        return_value=[],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.is_guest_user",
+        return_value=False,
+    )
+    return dataset
+
+
+def _clickhouse(mocker: MockerFixture, dataset: SqlaTable) -> None:
+    # Imported lazily: clickhouse.py touches app.config at import time.
+    from superset.db_engine_specs.clickhouse import ClickHouseEngineSpec
+
+    mocker.patch.object(
+        SqlaTable, "db_engine_spec", new=property(lambda self: ClickHouseEngineSpec)
+    )
+
+
+def _filter_query(filters: list[dict[str, Any]]) -> QueryObjectDict:
+    return cast(
+        QueryObjectDict,
+        {
+            "granularity": None,
+            "from_dttm": None,
+            "to_dttm": None,
+            "is_timeseries": False,
+            "groupby": ["city"],
+            "metrics": ["count"],
+            "filter": filters,
+            "columns": [],
+        },
+    )
+
+
+def _sql(dataset: SqlaTable, filters: list[dict[str, Any]]) -> str:
+    return dataset.get_query_str_extended(
+        _filter_query(filters), mutate=False
+    ).sql.lower()
+
+
+def test_contains_any_generates_hasany(mocker: MockerFixture, app: Flask) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "skills",
+                    "op": FilterOperator.CONTAINS_ANY.value,
+                    "val": ["Driver", "Cook"],
+                }
+            ],
+        )
+    assert "hasany(skills, array('driver', 'cook'))" in sql
+
+
+def test_contains_any_numeric_array_coerces_values(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """Values for a numeric array must render as numbers, not quoted strings."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "scores",
+                    "op": FilterOperator.CONTAINS_ANY.value,
+                    "val": ["5", "6"],
+                }
+            ],
+        )
+    assert "hasany(scores, array(5, 6))" in sql
+    assert "'5'" not in sql
+
+
+def test_equals_numeric_array_coerces_values(mocker: MockerFixture, app: Flask) -> None:
+    """Whole-array equality on a numeric array coerces the parsed literal."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "scores",
+                    "op": FilterOperator.EQUALS.value,
+                    "val": "[5, 6]",
+                }
+            ],
+        )
+    assert "scores = array(5, 6)" in sql
+
+
+def test_contains_all_generates_hasall(mocker: MockerFixture, app: Flask) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "skills",
+                    "op": FilterOperator.CONTAINS_ALL.value,
+                    "val": ["Driver", "Cook"],
+                }
+            ],
+        )
+    assert "hasall(skills, array('driver', 'cook'))" in sql
+
+
+def test_is_empty_generates_length_zero(mocker: MockerFixture, app: Flask) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(dataset, [{"col": "skills", "op": FilterOperator.IS_EMPTY.value}])
+    assert "length(skills) = 0" in sql
+
+
+def test_is_not_empty_generates_length_gt_zero(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset, [{"col": "skills", "op": FilterOperator.IS_NOT_EMPTY.value}]
+        )
+    assert "length(skills) > 0" in sql
+
+
+def test_contains_resolves_to_hasany(mocker: MockerFixture, app: Flask) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "skills",
+                    "op": FilterOperator.CONTAINS_ANY.value,
+                    "val": ["Driver"],
+                }
+            ],
+        )
+    assert "hasany(skills" in sql
+
+
+def test_element_op_on_scalar_column_raises(mocker: MockerFixture, app: Flask) -> None:
+    """CONTAINS_ANY on a scalar column is rejected on an array-capable engine."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():  # noqa: SIM117
+        with pytest.raises(QueryObjectValidationError):
+            _sql(
+                dataset,
+                [
+                    {
+                        "col": "city",
+                        "op": FilterOperator.CONTAINS_ANY.value,
+                        "val": ["NYC"],
+                    }
+                ],
+            )
+
+
+def test_element_op_unsupported_engine_raises(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """On an engine without array support (sqlite) the array op is rejected."""
+    dataset = _make_dataset(mocker)
+    with app.test_request_context():  # noqa: SIM117
+        with pytest.raises(QueryObjectValidationError):
+            _sql(dataset, [{"col": "skills", "op": FilterOperator.IS_EMPTY.value}])
+
+
+def test_equals_on_array_parses_literal(mocker: MockerFixture, app: Flask) -> None:
+    """A pasted array literal for = is parsed into col = array(...)."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "skills",
+                    "op": FilterOperator.EQUALS.value,
+                    "val": "['Driver', 'Cook']",
+                }
+            ],
+        )
+    assert "skills = array('driver', 'cook')" in sql
+
+
+def test_equals_on_array_plain_value_fallback(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    """A plain (non-bracketed) value becomes a single-element array."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [{"col": "skills", "op": FilterOperator.EQUALS.value, "val": "Driver"}],
+        )
+    assert "skills = array('driver')" in sql
+
+
+def test_in_on_array_parses_literals(mocker: MockerFixture, app: Flask) -> None:
+    """Whole-array IN parses each pasted array literal into its own array."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(
+            dataset,
+            [
+                {
+                    "col": "skills",
+                    "op": FilterOperator.IN.value,
+                    "val": ["['Driver']", "['Cook']"],
+                }
+            ],
+        )
+    assert "skills in (array('driver'), array('cook'))" in sql
@@ -0,0 +1,130 @@
+# 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.
+"""The Length array filter operators: length(col) compared to a number."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from flask import Flask
+from pytest_mock import MockerFixture
+
+from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn
+from superset.exceptions import QueryObjectValidationError
+from superset.models.core import Database
+from superset.superset_typing import QueryObjectDict
+from superset.utils.core import FilterOperator
+
+
+def _make_dataset(mocker: MockerFixture) -> SqlaTable:
+    database = Database(id=1, database_name="test_db", sqlalchemy_uri="sqlite://")
+    columns = [
+        TableColumn(column_name="skills", type="Array(String)"),
+        TableColumn(column_name="city", type="VARCHAR(100)"),
+    ]
+    dataset = SqlaTable(
+        table_name="jobs",
+        columns=columns,
+        database=database,
+        metrics=[SqlMetric(metric_name="count", expression="COUNT(*)")],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
+        return_value=[],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.is_guest_user",
+        return_value=False,
+    )
+    return dataset
+
+
+def _clickhouse(mocker: MockerFixture, dataset: SqlaTable) -> None:
+    from superset.db_engine_specs.clickhouse import ClickHouseEngineSpec
+
+    mocker.patch.object(
+        SqlaTable, "db_engine_spec", new=property(lambda self: ClickHouseEngineSpec)
+    )
+
+
+def _sql(dataset: SqlaTable, op: str, val: Any, col: str = "skills") -> str:
+    query: QueryObjectDict = {
+        "granularity": None,
+        "from_dttm": None,
+        "to_dttm": None,
+        "is_timeseries": False,
+        "groupby": ["city"],
+        "metrics": ["count"],
+        "filter": [{"col": col, "op": op, "val": val}],
+        "columns": [],
+    }
+    return dataset.get_query_str_extended(query, mutate=False).sql.lower()
+
+
+@pytest.mark.parametrize(
+    "op,expected",
+    [
+        (FilterOperator.LENGTH_EQUALS, "length(skills) = 3"),
+        (FilterOperator.LENGTH_GREATER_THAN, "length(skills) > 3"),
+        (FilterOperator.LENGTH_LESS_THAN, "length(skills) < 3"),
+        (FilterOperator.LENGTH_GREATER_THAN_OR_EQUALS, "length(skills) >= 3"),
+        (FilterOperator.LENGTH_LESS_THAN_OR_EQUALS, "length(skills) <= 3"),
+    ],
+)
+def test_length_operators_generate_length_comparison(
+    mocker: MockerFixture, app: Flask, op: FilterOperator, expected: str
+) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(dataset, op.value, 3)
+    assert expected in sql
+
+
+def test_length_accepts_string_number(mocker: MockerFixture, app: Flask) -> None:
+    """A numeric string value is coerced (e.g. '2' -> length(col) > 2)."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():
+        sql = _sql(dataset, FilterOperator.LENGTH_GREATER_THAN.value, "2")
+    assert "length(skills) > 2" in sql
+
+
+def test_length_non_numeric_value_raises(mocker: MockerFixture, app: Flask) -> None:
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():  # noqa: SIM117
+        with pytest.raises(QueryObjectValidationError):
+            _sql(dataset, FilterOperator.LENGTH_EQUALS.value, "abc")
+
+
+def test_length_on_scalar_column_raises(mocker: MockerFixture, app: Flask) -> None:
+    """Length on a scalar column is rejected even on an array-capable engine."""
+    dataset = _make_dataset(mocker)
+    _clickhouse(mocker, dataset)
+    with app.test_request_context():  # noqa: SIM117
+        with pytest.raises(QueryObjectValidationError):
+            _sql(dataset, FilterOperator.LENGTH_GREATER_THAN.value, 1, col="city")
+
+
+def test_length_unsupported_engine_raises(mocker: MockerFixture, app: Flask) -> None:
+    """On an engine without array support (sqlite) the length op is rejected."""
+    dataset = _make_dataset(mocker)
+    with app.test_request_context():  # noqa: SIM117
+        with pytest.raises(QueryObjectValidationError):
+            _sql(dataset, FilterOperator.LENGTH_GREATER_THAN.value, 1)
@@ -14,7 +14,13 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from superset.utils.pandas_postprocessing import escape_separator, unescape_separator
+import inspect
+
+from superset.utils.pandas_postprocessing import (
+    escape_separator,
+    pivot,
+    unescape_separator,
+)


 def test_escape_separator():
@@ -28,3 +34,19 @@ def test_escape_separator():
    escape_string = escape_separator("hello,world")
    assert escape_string == r"hello\,world"
    assert unescape_separator(escape_string) == "hello,world"
+
+
+def test_validate_column_args_preserves_signature():
+    """
+    The decorator must not hide the signature of the operation it wraps.
+
+    `inspect.signature` follows `__wrapped__`, which `functools.wraps` sets.
+    Without it every decorated operation reports `(df, **options)`, and code
+    that inspects the signature -- see `QueryObject._drop_unsupported_options`
+    -- cannot tell a supported option from an unsupported one.
+    """
+    parameters = inspect.signature(pivot).parameters
+
+    assert pivot.__name__ == "pivot"
+    assert "options" not in parameters
+    assert {"index", "aggregates", "columns"} <= set(parameters)
@@ -22,6 +22,7 @@ from superset.common.query_object import QueryObject
 from superset.connectors.sqla.models import SqlaTable
 from superset.models.core import Database
 from superset.superset_typing import Metric
+from superset.utils import pandas_postprocessing
 from superset.utils.core import override_user


@@ -438,3 +439,143 @@ def test_cache_key_cache_impersonation_on_with_different_user_and_db_impersonati
        ],
        any_order=True,
    )
+
+
+def test_post_processing_drops_unsupported_options():
+    """
+    An option that the operation no longer accepts is dropped, not passed on.
+
+    A chart saved by an older version of Superset stores `flatten_columns` in
+    the options of its `pivot` operation. `pivot` lost that parameter when
+    flattening became its own operation, so replaying the stored query_context
+    raised `TypeError: pivot() got an unexpected keyword argument
+    'flatten_columns'`.
+    """
+    query_object = QueryObject(
+        row_limit=1,
+        post_processing=[
+            {
+                "operation": "pivot",
+                "options": {
+                    "index": ["__timestamp"],
+                    "columns": ["genre"],
+                    "aggregates": {"count": {"operator": "mean"}},
+                    "drop_missing_columns": False,
+                    "flatten_columns": True,
+                    "reset_index": True,
+                },
+            }
+        ],
+    )
+
+    options = query_object.post_processing[0]["options"]
+    assert "flatten_columns" not in options
+    assert "reset_index" not in options
+    assert options["drop_missing_columns"] is False
+    assert options["index"] == ["__timestamp"]
+
+
+def test_post_processing_keeps_supported_options():
+    """Options the operation accepts are left alone."""
+    post_processing = [
+        {
+            "operation": "pivot",
+            "options": {"index": ["__timestamp"], "aggregates": {}},
+        }
+    ]
+    query_object = QueryObject(row_limit=1, post_processing=post_processing)
+
+    assert query_object.post_processing == post_processing
+
+
+def test_post_processing_keeps_unknown_operation():
+    """
+    An unknown operation is kept, so that `exec_post_processing` can report it
+    as an `InvalidPostProcessingError` rather than being silently dropped here.
+    """
+    query_object = QueryObject(
+        row_limit=1,
+        post_processing=[{"operation": "does_not_exist", "options": {"a": 1}}, None],
+    )
+
+    assert query_object.post_processing == [
+        {"operation": "does_not_exist", "options": {"a": 1}}
+    ]
+
+
+def test_post_processing_drops_the_dataframe_parameter():
+    """
+    The DataFrame parameter is not an option.
+
+    `exec_post_processing` calls `operation(df, **options)`, so an option named
+    after the first parameter would raise `TypeError: pivot() got multiple
+    values for argument 'df'`.
+    """
+    query_object = QueryObject(
+        row_limit=1,
+        post_processing=[
+            {
+                "operation": "pivot",
+                "options": {"df": "malformed", "index": ["a"], "aggregates": {}},
+            }
+        ],
+    )
+
+    options = query_object.post_processing[0]["options"]
+    assert "df" not in options
+    assert options["index"] == ["a"]
+
+
+def test_post_processing_keeps_options_of_a_variadic_operation():
+    """An operation that accepts `**kwargs` accepts every option."""
+
+    def variadic(df, **kwargs):
+        return df
+
+    post_processing = [{"operation": "variadic", "options": {"anything": 1}}]
+    with patch.object(pandas_postprocessing, "variadic", variadic, create=True):
+        query_object = QueryObject(row_limit=1, post_processing=post_processing)
+
+    assert query_object.post_processing == post_processing
+
+
+def test_post_processing_drops_a_variadic_positional_option():
+    """
+    A `*args` parameter cannot be filled by a keyword argument.
+
+    `exec_post_processing` calls the operation as `operation(df, **options)`,
+    so an option named after a `*args` parameter would raise `TypeError:
+    variadic_positional() got an unexpected keyword argument 'args'` even
+    though the name appears in the signature.
+    """
+
+    def variadic_positional(df, *args, index=None):  # pylint: disable=unused-argument
+        return df
+
+    with patch.object(
+        pandas_postprocessing, "variadic_positional", variadic_positional, create=True
+    ):
+        query_object = QueryObject(
+            row_limit=1,
+            post_processing=[
+                {
+                    "operation": "variadic_positional",
+                    "options": {"args": [1], "index": ["a"]},
+                }
+            ],
+        )
+
+    options = query_object.post_processing[0]["options"]
+    assert "args" not in options
+    assert options["index"] == ["a"]
+
+
+def test_post_processing_keeps_an_entry_without_an_operation():
+    """
+    An entry that names no operation is kept, so that `exec_post_processing`
+    reports it as an `InvalidPostProcessingError`.
+    """
+    post_processing = [{"options": {"a": 1}}]
+    query_object = QueryObject(row_limit=1, post_processing=post_processing)
+
+    assert query_object.post_processing == post_processing
Author	SHA1	Message	Date
rusackasandClaude Opus 4.8	08cb0f6c15	fix(lineage): redact chart dataset details when caller lacks access Mirrors the dashboard lineage endpoint's gating: schema/table/database name are only exposed when the caller can access the underlying datasource, so the chart endpoint no longer leaks that info unconditionally. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:25:38 -07:00
EvanandClaude Opus 4.8	effc239721	fix(lineage): nullable schema fields, lazy modal fetch, N+1 + type hints - Mark database_name/slug as allow_none in lineage response schemas to match the nullable values the endpoints actually return. - Defer LineageModal's lineage fetch until the modal opens (beforeOpen + the hooks' skip flag) so rendering it inside a dropdown no longer fires the endpoint on menu open. - selectinload Slice.dashboards / Dashboard.slices in get_related_objects to avoid N+1 queries when building dataset lineage. - Drop the unreachable dataset branch in getEntityUrl (no standalone dataset page) and add type hints to the new lineage locals. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:46 -07:00
EvanandClaude Opus 4.8	18bb76e885	fix(lineage): address review feedback on schema, security, and UI - mark redacted dashboard lineage dataset fields nullable in the schema - filter dataset lineage chart_ids by chart access - responsive Sankey width and fallback dataset->dashboard edges - empty state for unsaved datasets; i18n legend labels; typed test mocks Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:21 -07:00
EvanandClaude Opus 4.8	dad323a394	fix(lineage): skip dataset lineage fetch when id is missing Pass '' instead of 0 as the fallback so isEmptyId skips the request rather than calling /api/v1/dataset/0/lineage. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:21 -07:00
EvanandClaude Opus 4.8	7c1991ac55	fix(lineage): tighten types on LineageView, drop any usage Replace `Record<string, any>` on additionalInfo with an explicit value union, and type the Sankey node click handler params instead of `any`. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:21 -07:00
EvanandClaude Opus 4.8	8d78b74e86	test(lineage): add coverage for the skip parameter in api resource hooks Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:20 -07:00
EvanandClaude Opus 4.8	b9921ccc9d	test(lineage): add unit tests for lineage API hooks Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:20 -07:00
EvanandClaude Opus 4.8	c0af873a9c	fix(lineage): address review feedback (i18n, permissions, node identity) - Wrap untranslated "Details" JSX text in t() (fixes pre-commit custom rules + babel-extract regression) - Filter dataset/chart/dashboard lineage endpoints by the current user's permissions so they never expose charts, dashboards, or datasource metadata the user cannot access (mirrors existing related_objects / get_datasets redaction patterns) - Register DashboardLineageResponseSchema in openapi_spec_component_schemas to resolve the dangling $ref in the generated API spec - Key Sankey graph nodes and the node-details map by a stable unique identity (type:id) so entities sharing a display name no longer collapse into a single node; keep the human-readable title as a separate label - Add a skip mechanism to useApiV1Resource / lineage hooks so empty-id callers (e.g. LineageModal) no longer fire requests against invalid endpoints like /api/v1/chart//lineage - Defer the dataset edit page lineage fetch until the Lineage tab is active - Make "View lineage" available in dashboard view mode, not only edit mode - Compare data["result"] in lineage integration tests to match the result-wrapped API contract Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:20 -07:00
Claude CodeandJonathan Alberth Quispe Fuentes	2ab7b388d0	feat(lineage): add lineage visualization across datasets, charts and dashboards Adopts #37701 by @qf-jonathan. Adds interactive Sankey-based lineage that shows the data relationships for a dataset, chart, or dashboard: three REST endpoints (/api/v1/lineage/{dataset\|chart\|dashboard}/<id>) plus a LineageView/LineageModal frontend surfaced from the dataset editor and the chart/dashboard menus. Rebased onto current master and updated for drift since the original branch: - repointed lineage imports from the removed '@apache-superset/core/ui' to '@apache-superset/core/translation' and '/theme' - re-applied the dataset lineage tab onto the migrated (TypeScript) DatasourceEditor Closes #37701 Co-authored-by: Jonathan Alberth Quispe Fuentes <qf.jonathan@gmail.com> Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 21:18:19 -07:00
Arya Ketan Claude Opus 5 Evan Rusackas	f9cedf84e2	fix: drop post-processing options the operation no longer accepts (#42927 ) Signed-off-by: Arya Ketan <aryaketan@sharechat.co> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Evan Rusackas <evan@preset.io>	2026-08-19 18:15:15 -07:00
Grégoire GaillyandEvan Rusackas	c2d653b4b8	fix: set maxHeight of List components to height when in AutoSizer (#43056 ) Co-authored-by: Evan Rusackas <evan@preset.io>	2026-08-19 16:56:30 -07:00
Đỗ Trọng HảiandJoe Li	5a96c3f538	chore(ci): disable Git commit info capture in Playwright E2E tests to avoid timeout (#43213 ) Signed-off-by: hainenber <dotronghai96@gmail.com> Co-authored-by: Joe Li <joe@preset.io>	2026-08-19 16:54:44 -07:00
ʈᵃᵢ	faf7c34c0a	fix(explore): legacy boolean filters and limit available operators based on calculated column type (#43341 )	2026-08-19 15:37:09 -07:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	b8fca2145d	chore(deps): bump astral-sh/setup-uv from 9.0.0 to 10.0.0 (#43322 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 14:40:02 -07:00
Alejandro Solares	c10054f521	fix(plugin-chart-chord): declare react as a peerDependency (#43304 )	2026-08-19 17:35:38 -04:00
Amin GhadersohiandClaude	8c500ccee1	fix(users): show password validation errors (#43191 ) Co-authored-by: Claude <noreply@anthropic.com>	2026-08-19 16:57:46 -04:00
Joe LiandClaude Sonnet 5	6d77efad29	fix(chart): stop contextmenu propagation in BigNumberViz (#43267 ) Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>	2026-08-19 13:57:28 -07:00
Amin GhadersohiandClaude	8222db3340	fix(dataset): preserve legacy default dashboard URLs (#43190 ) Co-authored-by: Claude <noreply@anthropic.com>	2026-08-19 14:44:54 -04:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	01ce8358a6	chore(deps-dev): bump globals from 17.9.0 to 17.10.0 in /superset-websocket (#43321 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 11:36:48 -07:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	53a8a0e140	chore(deps): bump the docusaurus-openapi group in /docs with 2 updates (#43323 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 11:36:44 -07:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	eafbff9f8d	chore(deps-dev): bump globals from 17.9.0 to 17.10.0 in /docs (#43324 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 11:36:37 -07:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	1339bcd9da	chore(deps): bump swagger-ui-react from 5.32.12 to 5.32.13 in /docs (#43325 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 11:36:34 -07:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	334e280489	chore(deps): bump dompurify from 3.4.12 to 3.4.13 in /superset-frontend (#43326 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 11:36:30 -07:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	c07f3ebf2d	chore(deps-dev): bump @swc/plugin-emotion from 14.15.0 to 14.19.0 in /superset-frontend (#43328 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-08-19 11:36:26 -07:00
Sumit KumarandClaude Opus 4.8	1569915096	feat(multi-value): array-typed column filters with two-tier operators (ClickHouse MVP) (#41279 ) Signed-off-by: thedeceptio <thedeceptio@gmail.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>	2026-08-19 10:56:49 -07:00
BexultanandBexultan Mustafin	fde0ba26d1	fix(mcp): validate virtual dataset metadata and surface errors (#43129 ) Co-authored-by: Bexultan Mustafin <bexultan.mustafin@ffins.kz>	2026-08-19 10:50:37 -07:00