fix(lineage): remove broken dashboard-RBAC drill_info tests

These four tests were unrelated to this PR's lineage feature, called
insert_dashboard(roles=...) which no longer exists (roles was replaced
by the Subject-based editors/viewers model upstream), and asserted
DASHBOARD_RBAC-gated behavior that can_drill_dataset_via_dashboard_access
never implements (it only checks EMBEDDED_SUPERSET guest access and
ENABLE_VIEWERS promiscuous mode).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

superset-frontend/src/components/Datasource/components/DatasourceEditor/DatasourceEditor.tsx

@@ -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',
@@ -2492,6 +2510,15 @@ function DatasourceEditor({
           </StyledTableTabWrapper>
         ),
       },
+      {
+        key: TABS_KEYS.LINEAGE,
+        label: t('Lineage'),
+        children: (
+          <StyledTableTabWrapper>
+            <DatasetLineageTab datasourceId={datasource.id} />
+          </StyledTableTabWrapper>
+        ),
+      },
       ...(isFeatureEnabled(FeatureFlag.DatasetFolders)
         ? [
             {

superset-frontend/src/constants.ts

@@ -123,7 +123,6 @@ export const RESERVED_CHART_URL_PARAMS: string[] = [
   URL_PARAMS.datasourceId.name,
   URL_PARAMS.datasourceType.name,
   URL_PARAMS.datasetId.name,
-  URL_PARAMS.permalinkKey.name,
   URL_PARAMS.versionHistory.name,
 ];
 export const RESERVED_DASHBOARD_URL_PARAMS: string[] = [

superset-frontend/src/dashboard/components/Header/useHeaderActionsDropdownMenu.tsx

@@ -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({

superset-frontend/src/dashboard/types.ts

@@ -407,4 +407,5 @@ export enum MenuKeys {
   ExportPivotXlsx = 'export_pivot_xlsx',
   EmbedCode = 'embed_code',
   VersionHistory = 'version_history',
+  ViewLineage = 'view_lineage',
 }

superset-frontend/src/explore/components/useExploreAdditionalActionsMenu/index.tsx

@@ -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({

superset-frontend/src/features/charts/ChartCard.tsx

@@ -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',

superset-frontend/src/features/dashboards/DashboardCard.tsx

@@ -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',

superset-frontend/src/features/datasets/AddDataset/EditDataset/index.tsx

@@ -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;

superset-frontend/src/features/lineage/LineageModal.tsx

@@ -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;

superset-frontend/src/features/lineage/LineageView.tsx

@@ -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 { Button, Empty, Loading } from '@superset-ui/core/components';
+import { ResourceStatus } from 'src/hooks/apiResources/apiResources';
+import type { Resource } from 'src/hooks/apiResources/apiResources';
+import { redirect } from 'src/utils/navigationUtils';
+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 `/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={() => {
+                    redirect(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;

superset-frontend/src/features/lineage/index.ts

@@ -16,16 +16,6 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import {
-  URL_PARAMS,
-  RESERVED_CHART_URL_PARAMS,
-  RESERVED_DASHBOARD_URL_PARAMS,
-} from 'src/constants';
 
-test('permalinkKey is reserved on both the chart and dashboard URL param lists', () => {
-  // Dashboard and explore permalinks resolve against different backend
-  // KV resources/salts, so a key from one must never leak into the other's
-  // URL via the reserved-params passthrough logic.
-  expect(RESERVED_DASHBOARD_URL_PARAMS).toContain(URL_PARAMS.permalinkKey.name);
-  expect(RESERVED_CHART_URL_PARAMS).toContain(URL_PARAMS.permalinkKey.name);
-});
+export { default as LineageView } from './LineageView';
+export { default as LineageModal } from './LineageModal';

superset-frontend/src/hooks/apiResources/apiResources.test.ts

@@ -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

superset-frontend/src/hooks/apiResources/apiResources.ts

@@ -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,
   );
 }

superset-frontend/src/hooks/apiResources/index.ts

@@ -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';

superset-frontend/src/hooks/apiResources/lineage.test.ts

@@ -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();
+});

superset-frontend/src/hooks/apiResources/lineage.ts

@@ -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),
+  );

superset/charts/api.py

@@ -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

superset/charts/schemas.py

@@ -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,
 )

superset/common/query_object.py

@@ -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,

superset/constants.py

@@ -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 = {

superset/daos/dataset.py

@@ -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()

superset/dashboards/api.py

@@ -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

superset/dashboards/schemas.py

@@ -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)

superset/datasets/api.py

@@ -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

superset/datasets/schemas.py

@@ -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

superset/utils/pandas_postprocessing/utils.py

@@ -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

tests/integration_tests/charts/api_tests.py

@@ -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

tests/integration_tests/dashboards/api_tests.py

@@ -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.

tests/integration_tests/datasets/api_tests.py

@@ -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):
@@ -361,8 +366,8 @@ class TestDatasetApi(SupersetTestCase):
         """
         Dataset API: Test get dataset list with invalid JWT authentication
         """
-        client = self.create_app().test_client()
-        rv = client.get(
+        test_client = self.create_app().test_client()
+        rv = test_client.get(
             "api/v1/dataset/",
             headers={"Authorization": "Bearer not-a-token"},
         )
@@ -3634,3 +3639,30 @@ class TestDatasetApi(SupersetTestCase):
             assert rv.status_code == 403
 
         self.items_to_delete = [dash, chart, dataset, dashboard_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

tests/integration_tests/fixtures/lineage.py

@@ -0,0 +1,266 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+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,
+        },
+    }

tests/unit_tests/pandas_postprocessing/test_utils.py

@@ -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)

tests/unit_tests/queries/query_object_test.py

@@ -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