Fix lint

Co-authored-by: Claude <noreply@anthropic.com>

superset-frontend/src/components/Chart/DrillDetail/DrillDetailPane.test.tsx

@@ -216,6 +216,12 @@ test('should render the error', async () => {
     .spyOn(SupersetClient, 'post')
     .mockRejectedValue(new Error('Something went wrong'));
   await waitForRender();
+  // The error is wrapped in an Alert component with a stable headline and the
+  // raw error text in the description — no more bare ``<pre>`` elements.
+  expect(await screen.findByRole('alert')).toBeVisible();
+  expect(
+    await screen.findByText('Failed to load drill-to-detail rows'),
+  ).toBeVisible();
   expect(screen.getByText('Error: Something went wrong')).toBeInTheDocument();
 });
 

superset-frontend/src/components/Chart/DrillDetail/DrillDetailPane.tsx

@@ -42,6 +42,7 @@ import BooleanCell from '@superset-ui/core/components/Table/cell-renderers/Boole
 import NullCell from '@superset-ui/core/components/Table/cell-renderers/NullCell';
 import TimeCell from '@superset-ui/core/components/Table/cell-renderers/TimeCell';
 import { EmptyState, Loading } from '@superset-ui/core/components';
+import { Alert } from '@apache-superset/core/components';
 import { getDatasourceSamples } from 'src/components/Chart/chartAction';
 import Table, {
   ColumnsType,
@@ -362,13 +363,18 @@ export default function DrillDetailPane({
   if (responseError) {
     // Render error if page download failed
     tableContent = (
-      <pre
+      <div
         css={css`
           margin-top: ${theme.sizeUnit * 4}px;
         `}
       >
-        {responseError}
-      </pre>
+        <Alert
+          type="error"
+          showIcon
+          message={t('Failed to load drill-to-detail rows')}
+          description={responseError}
+        />
+      </div>
     );
   } else if (bootstrapping) {
     // Render loading if first page hasn't loaded

superset-frontend/src/components/Chart/useDrillDetailMenuItems/index.tsx

@@ -50,6 +50,7 @@ const DISABLED_REASONS = {
   DATABASE: t(
     'Drill to detail is disabled for this database. Change the database settings to enable it.',
   ),
+  DATASOURCE: t('Drill to detail is not available for this datasource type.'),
   NO_AGGREGATIONS: t(
     'Drill to detail is disabled because this chart does not group data by dimension value.',
   ),
@@ -116,6 +117,17 @@ export const useDrillDetailMenuItems = ({
       datasources[formData.datasource]?.database?.disable_drill_to_detail,
   );
 
+  // Capability flag on the datasource itself. Datasources that don't model
+  // raw rows (e.g. semantic views) opt out via ``supports_drill_to_detail``
+  // in the explore data payload.
+  const datasourceSupportsDrillToDetail = useSelector<
+    RootState,
+    boolean | undefined
+  >(
+    ({ datasources }) =>
+      datasources[formData.datasource]?.supports_drill_to_detail,
+  );
+
   const openModal = useCallback(
     (filters: BinaryQueryObjectFilterClause[], event: MouseEvent) => {
       onClick(event);
@@ -158,7 +170,10 @@ export const useDrillDetailMenuItems = ({
 
   let drillDisabled;
   let drillByDisabled;
-  if (drillToDetailDisabled) {
+  if (datasourceSupportsDrillToDetail === false) {
+    drillDisabled = DISABLED_REASONS.DATASOURCE;
+    drillByDisabled = DISABLED_REASONS.DATASOURCE;
+  } else if (drillToDetailDisabled) {
     drillDisabled = DISABLED_REASONS.DATABASE;
     drillByDisabled = DISABLED_REASONS.DATABASE;
   } else if (handlesDimensionContextMenu) {

superset-frontend/src/components/Chart/useDrillDetailMenuItems/useDrillDetailMenuItems.test.tsx

@@ -444,3 +444,45 @@ test('context menu renders <NULL> for null dimension values', async () => {
   await expectDrillToDetailByEnabled();
   await expectDrillToDetailByDimension(filterNull);
 });
+
+const buildStateWithUnsupportedDatasource = () => {
+  const baseState = getMockStoreWithNativeFilters().getState();
+  const datasourceKey = defaultFormData.datasource as string;
+  return {
+    ...baseState,
+    datasources: {
+      ...baseState.datasources,
+      [datasourceKey]: {
+        ...baseState.datasources[datasourceKey],
+        supports_drill_to_detail: false,
+      },
+    },
+  };
+};
+
+test('dropdown menu when datasource opts out via supports_drill_to_detail=false', async () => {
+  cleanup();
+  render(<MockRenderChart formData={defaultFormData} />, {
+    useRouter: true,
+    useRedux: true,
+    initialState: buildStateWithUnsupportedDatasource(),
+  });
+
+  await expectDrillToDetailDisabled(
+    'Drill to detail is not available for this datasource type.',
+  );
+  await expectNoDrillToDetailBy();
+});
+
+test('context menu when datasource opts out via supports_drill_to_detail=false', async () => {
+  cleanup();
+  render(<MockRenderChart formData={defaultFormData} isContextMenu />, {
+    useRouter: true,
+    useRedux: true,
+    initialState: buildStateWithUnsupportedDatasource(),
+  });
+
+  const message = 'Drill to detail is not available for this datasource type.';
+  await expectDrillToDetailDisabled(message);
+  await expectDrillToDetailByDisabled(message);
+});

superset-frontend/src/dashboard/types.ts

@@ -232,6 +232,10 @@ export type Datasource = Dataset & {
   // Populated by the dashboard datasets API alongside ``type``; declared here
   // so callers can rely on structural typing instead of casting.
   datasource_type?: DatasourceType;
+  /** False when the datasource can't return row samples (e.g. semantic views). */
+  supports_samples?: boolean;
+  /** False when the datasource can't answer drill-to-detail requests. */
+  supports_drill_to_detail?: boolean;
 };
 export type DatasourcesState = {
   [key: string]: Datasource;

superset-frontend/src/explore/components/DataTablesPane/DataTablesPane.tsx

@@ -241,25 +241,44 @@ export const DataTablesPane = ({
     }
   }, [resultsTabFallback]);
 
+  // Hide the Samples tab for datasources that don't expose raw rows
+  // (e.g. semantic views). The check is intentionally ``=== false`` so that
+  // datasources from older backends that don't send the flag still show the
+  // tab and preserve current behavior.
+  const showSamplesTab = datasource?.supports_samples !== false;
+
+  // If the datasource swaps to one that doesn't support samples while the
+  // Samples tab is active (e.g. the user picks a semantic view), the tab
+  // disappears from ``tabItems`` and ``activeTabKey`` is orphaned. Fall back
+  // to Results so the panel keeps rendering content.
+  useEffect(() => {
+    if (!showSamplesTab && activeTabKey === ResultTypes.Samples) {
+      setActiveTabKey(ResultTypes.Results);
+    }
+  }, [showSamplesTab, activeTabKey]);
   const tabItems = [
     ...queryResultsPanes,
-    {
-      key: ResultTypes.Samples,
-      label: t('Samples'),
-      children: (
-        <StyledDiv>
-          <SamplesPane
-            datasource={datasource}
-            queryFormData={queryFormData}
-            queryForce={queryForce}
-            isRequest={isRequest.samples}
-            setForceQuery={setForceQuery}
-            isVisible={ResultTypes.Samples === activeTabKey}
-            canDownload={canDownload}
-          />
-        </StyledDiv>
-      ),
-    },
+    ...(showSamplesTab
+      ? [
+          {
+            key: ResultTypes.Samples,
+            label: t('Samples'),
+            children: (
+              <StyledDiv>
+                <SamplesPane
+                  datasource={datasource}
+                  queryFormData={queryFormData}
+                  queryForce={queryForce}
+                  isRequest={isRequest.samples}
+                  setForceQuery={setForceQuery}
+                  isVisible={ResultTypes.Samples === activeTabKey}
+                  canDownload={canDownload}
+                />
+              </StyledDiv>
+            ),
+          },
+        ]
+      : []),
   ];
 
   return (

superset-frontend/src/explore/components/DataTablesPane/components/SamplesPane.tsx

@@ -21,6 +21,7 @@ import { t } from '@apache-superset/core/translation';
 import { ensureIsArray } from '@superset-ui/core';
 import { datasetLabelLower } from 'src/features/semanticLayers/label';
 import { styled } from '@apache-superset/core/theme';
+import { Alert } from '@apache-superset/core/components';
 import { EmptyState, Loading } from '@superset-ui/core/components';
 import { GenericDataType } from '@apache-superset/core/common';
 import { GridTable } from 'src/components/GridTable';
@@ -35,7 +36,7 @@ import {
 import { TableControls, ROW_LIMIT_OPTIONS } from './DataTableControls';
 import { SamplesPaneProps } from '../types';
 
-const Error = styled.pre`
+const ErrorAlertWrapper = styled.div`
   margin-top: ${({ theme }) => `${theme.sizeUnit * 4}px`};
 `;
 
@@ -155,7 +156,14 @@ export const SamplesPane = ({
           rowLimitOptions={ROW_LIMIT_OPTIONS}
           onRowLimitChange={handleRowLimitChange}
         />
-        <Error>{responseError}</Error>
+        <ErrorAlertWrapper>
+          <Alert
+            type="error"
+            showIcon
+            message={t('Failed to load samples')}
+            description={responseError}
+          />
+        </ErrorAlertWrapper>
       </>
     );
   }

superset-frontend/src/explore/components/DataTablesPane/components/useResultsPane.tsx

@@ -27,13 +27,14 @@ import {
   QueryData,
 } from '@superset-ui/core';
 import { styled } from '@apache-superset/core/theme';
+import { Alert } from '@apache-superset/core/components';
 import { EmptyState, Loading } from '@superset-ui/core/components';
 import { getChartDataRequest } from 'src/components/Chart/chartAction';
 import { ResultsPaneProps, QueryResultInterface } from '../types';
 import { SingleQueryResultPane } from './SingleQueryResultPane';
 import { TableControls, ROW_LIMIT_OPTIONS } from './DataTableControls';
 
-const Error = styled.pre`
+const ErrorAlertWrapper = styled.div`
   margin-top: ${({ theme }) => `${theme.sizeUnit * 4}px`};
 `;
 
@@ -199,7 +200,14 @@ export const useResultsPane = ({
           isLoading={false}
           canDownload={canDownload}
         />
-        <Error>{responseError}</Error>
+        <ErrorAlertWrapper>
+          <Alert
+            type="error"
+            showIcon
+            message={t('Failed to load results')}
+            description={responseError}
+          />
+        </ErrorAlertWrapper>
       </>
     );
     return Array(queryCount).fill(err);

superset-frontend/src/explore/components/DataTablesPane/test/DataTablesPane.test.tsx

@@ -19,7 +19,12 @@
 import fetchMock from 'fetch-mock';
 import { FeatureFlag } from '@superset-ui/core';
 import * as copyUtils from 'src/utils/copy';
-import { render, screen, userEvent } from 'spec/helpers/testing-library';
+import {
+  render,
+  screen,
+  userEvent,
+  waitFor,
+} from 'spec/helpers/testing-library';
 import { setupAGGridModules } from '@superset-ui/core/components/ThemedAgGridReact';
 import { setItem, LocalStorageKeys } from 'src/utils/localStorageHelpers';
 import { DataTablesPane } from '..';
@@ -89,6 +94,48 @@ describe('DataTablesPane', () => {
     expect(await screen.findByLabelText('Collapse data panel')).toBeVisible();
   });
 
+  test('Hides Samples tab when datasource opts out via supports_samples=false', async () => {
+    const props = createDataTablesPaneProps(0);
+    const propsWithoutSamples = {
+      ...props,
+      datasource: { ...props.datasource, supports_samples: false },
+    };
+    render(<DataTablesPane {...propsWithoutSamples} />, { useRedux: true });
+    expect(await screen.findByText('Results')).toBeVisible();
+    expect(screen.queryByText('Samples')).not.toBeInTheDocument();
+  });
+
+  test('Falls back to Results when active Samples tab disappears mid-session', async () => {
+    // Regression for codeant Major finding on PR #41509: a datasource swap
+    // that hides the Samples tab while it was the active tab used to leave
+    // ``activeTabKey === 'samples'`` orphaned, rendering a blank panel.
+    const props = createDataTablesPaneProps(0);
+    const { rerender } = render(<DataTablesPane {...props} />, {
+      useRedux: true,
+    });
+
+    // Open the panel and pick the Samples tab.
+    userEvent.click(screen.getByLabelText('Expand data panel'));
+    userEvent.click(await screen.findByText('Samples'));
+    expect(await screen.findByLabelText('Collapse data panel')).toBeVisible();
+
+    // Swap to a datasource that doesn't support samples (e.g. a semantic
+    // view). The Samples tab should disappear and the panel should land on
+    // Results with content still rendered.
+    rerender(
+      <DataTablesPane
+        {...props}
+        datasource={{ ...props.datasource, supports_samples: false }}
+      />,
+    );
+    await waitFor(() => {
+      expect(screen.queryByText('Samples')).not.toBeInTheDocument();
+    });
+    expect(screen.getByText('Results')).toBeVisible();
+    // Panel stays expanded and renders Results content rather than going blank.
+    expect(screen.getByLabelText('Collapse data panel')).toBeVisible();
+  });
+
   test('Should copy data table content correctly', async () => {
     fetchMock.post(
       'glob:*/api/v1/chart/data?form_data=%7B%22slice_id%22%3A456%7D',

superset-frontend/src/explore/components/DataTablesPane/test/SamplesPane.test.tsx

@@ -84,10 +84,14 @@ describe('SamplesPane', () => {
     const props = createSamplesPaneProps({
       datasourceId: 36,
     });
-    const { findByText } = render(<SamplesPane {...props} />, {
+    const { findByText, findByRole } = render(<SamplesPane {...props} />, {
       useRedux: true,
     });
 
+    // The error is now rendered inside an Alert component, with a clear
+    // headline message and the raw error text as the description.
+    expect(await findByRole('alert')).toBeVisible();
+    expect(await findByText('Failed to load samples')).toBeVisible();
     expect(await findByText('Error: Bad request')).toBeVisible();
   });
 

superset-frontend/src/explore/types.ts

@@ -74,6 +74,18 @@ export type Datasource = Dataset & {
   schema?: string;
   is_sqllab_view?: boolean;
   extra?: string | object;
+  /**
+   * False when the datasource (e.g. a semantic view) doesn't model raw rows
+   * and therefore can't return a row sample. Defaults to true on the server
+   * side; missing here means the explore UI keeps current behavior.
+   */
+  supports_samples?: boolean;
+  /**
+   * False when the datasource doesn't model raw rows and therefore can't
+   * answer a drill-to-detail query. Tracked separately from
+   * ``supports_samples`` so the two capabilities can diverge.
+   */
+  supports_drill_to_detail?: boolean;
 };
 
 export interface ExplorePageInitialData {

superset/common/query_actions.py

@@ -235,6 +235,15 @@ def _get_drill_detail(
     # todo(yongjie): Remove this function,
     #  when determining whether samples should be applied to the time filter.
     datasource = _get_datasource(query_context, query_obj)
+    # Refuse for datasource types that don't model raw rows (e.g. semantic
+    # views). Mirrors the ``supports_samples`` gate on the ``/samples``
+    # endpoint so drill-detail is hard-blocked on the backend, not just
+    # hidden in the frontend menu. Defaults to ``True`` for any datasource
+    # class that doesn't explicitly opt out.
+    if not getattr(datasource, "supports_drill_to_detail", True):
+        raise QueryObjectValidationError(
+            _("Drill to detail is not available for this datasource type.")
+        )
     query_obj = copy.copy(query_obj)
     query_obj.is_timeseries = False
     query_obj.metrics = None

superset/connectors/sqla/models.py

@@ -193,6 +193,16 @@ class BaseDatasource(
     # Only some datasources support Row Level Security
     is_rls_supported: bool = False
 
+    # Datasources that can return raw row samples (anything backed by a SQL
+    # table can; semantic-layer abstractions cannot, since they only expose
+    # pre-defined metrics and dimensions).
+    supports_samples: bool = True
+
+    # Datasources that can answer "drill to detail" requests — i.e. fetch the
+    # raw rows underlying a chart cell. Conceptually similar to ``samples``
+    # but kept as a separate capability so the two can diverge.
+    supports_drill_to_detail: bool = True
+
     @property
     def name(self) -> str:
         # can be a Column or a property pointing to one
@@ -486,6 +496,8 @@ class BaseDatasource(
             "order_by_choices": self.order_by_choices,
             "verbose_map": self.verbose_map,
             "select_star": self.select_star,
+            "supports_samples": self.supports_samples,
+            "supports_drill_to_detail": self.supports_drill_to_detail,
         }
 
     def data_for_slices(  # pylint: disable=too-many-locals  # noqa: C901

superset/semantic_layers/models.py

@@ -200,6 +200,12 @@ class SemanticView(AuditMixinNullable, Model):
 
     __tablename__ = "semantic_views"
 
+    # Semantic views expose pre-defined metrics and dimensions, not raw rows,
+    # so neither the "Samples" tab in Explore nor the "Drill to detail"
+    # affordance from the chart 3-dots menu can return anything meaningful.
+    supports_samples: bool = False
+    supports_drill_to_detail: bool = False
+
     # Use integer as the primary key for cross-database auto-increment
     # compatibility (sa.Identity() is not supported in MySQL or SQLite).
     # The uuid column is a secondary unique identifier used in URLs and perms.
@@ -425,6 +431,8 @@ class SemanticView(AuditMixinNullable, Model):
             "sql": None,
             "select_star": None,
             "editors": [],
+            "supports_samples": self.supports_samples,
+            "supports_drill_to_detail": self.supports_drill_to_detail,
             "description": self.description,
             "table_name": self.name,
             "column_types": [

superset/superset_typing.py

@@ -346,6 +346,11 @@ class ExplorableData(TypedDict, total=False):
     always_filter_main_dttm: bool
     normalize_columns: bool
     rls_filters: list[dict[str, Any]]
+    # Set by datasources that cannot return raw row samples (e.g. semantic
+    # views, which only expose pre-defined metrics and dimensions).
+    supports_samples: bool
+    # Set by datasources that cannot answer drill-to-detail requests.
+    supports_drill_to_detail: bool
 
 
 VizData: TypeAlias = list[Any] | dict[Any, Any] | None

superset/utils/screenshot_utils.py

@@ -87,6 +87,29 @@ def resolve_screenshot_task_budget_seconds(
         return None
 
 
+# Fallback wall-clock budget, in seconds, for the entire tiled-screenshot
+# operation (element lookup plus all per-tile readiness/animation waits
+# combined), used when resolve_screenshot_task_budget_seconds() returns None
+# (no Celery task context -- e.g. synchronous thumbnail generation -- or no
+# usable task limit). The non-tiled readiness path treats None as "keep the
+# configured SCREENSHOT_LOAD_WAIT" because it makes exactly one bounded wait;
+# the tiled path cannot, because its per-tile waits accumulate: with N tiles,
+# an uncapped load_wait allows N * load_wait of total wall-clock time, so the
+# operation still needs one fixed total ceiling. Sized against the longest
+# Celery hard task_time_limit observed in production for report execution
+# (1740s), minus the same 300s cleanup margin the runtime derivation reserves
+# for combining tiles, building the PDF, and delivering the notification.
+TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS = 1440  # 1740s limit - 300s margin
+
+
+class ScreenshotTaskBudgetExceededError(RuntimeError):
+    """Raised when no safe task budget remains before screenshot capture."""
+
+
+class TiledScreenshotBudgetExceededError(ScreenshotTaskBudgetExceededError):
+    """Raised when the tiled-screenshot time budget runs out mid-capture."""
+
+
 try:
     from playwright.sync_api import TimeoutError as PlaywrightTimeout
 except ImportError:
@@ -251,7 +274,7 @@ def combine_screenshot_tiles(screenshot_tiles: list[bytes]) -> bytes:
         return screenshot_tiles[0]
 
 
-def take_tiled_screenshot(
+def take_tiled_screenshot(  # noqa: C901
     page: "Page",
     element_name: str,
     tile_height: int,
@@ -274,6 +297,12 @@ def take_tiled_screenshot(
 
     Returns:
         Combined screenshot bytes or None if failed
+
+    Raises:
+        TiledScreenshotBudgetExceededError: If the total time budget for the
+            tiled-screenshot operation runs out before every tile has been
+            verifiably captured. Callers must treat this as a hard failure
+            rather than fall back to an unchecked/partial screenshot.
     """
     context_suffix = f" [{log_context}]" if log_context else ""
     # Set right before re-raising the per-tile readiness timeout below, and
@@ -286,6 +315,15 @@ def take_tiled_screenshot(
     # match `except PlaywrightTimeout` and incorrectly propagate instead of
     # degrading to `None` like every other unexpected error in this function.
     readiness_timeout = False
+    # Cap the whole tiled operation against the running Celery task's own
+    # time limit, using the same runtime derivation as the non-tiled
+    # readiness wait (#42253/#42427). Unlike that path, a None budget does
+    # not mean "keep the configured timeout": per-tile waits accumulate, so
+    # the operation falls back to a fixed total ceiling instead.
+    wait_budget_seconds = resolve_screenshot_task_budget_seconds(log_context)
+    if wait_budget_seconds is None:
+        wait_budget_seconds = float(TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS)
+    start_time = time.monotonic()
     try:
         # Get the target element
         element = page.locator(f".{element_name}")
@@ -320,9 +358,44 @@ def take_tiled_screenshot(
         num_tiles = max(1, (dashboard_height + tile_height - 1) // tile_height)
         logger.info("Taking %s screenshot tiles", num_tiles)
 
-        screenshot_tiles = []
+        screenshot_tiles: list[bytes] = []
+
+        def _raise_if_budget_exhausted(elapsed: float, remaining_budget: float) -> None:
+            if remaining_budget > 0:
+                return
+            # A customer-side chart-loading issue (a slow/hung dashboard),
+            # not a Superset system fault, so this is a WARNING rather
+            # than an ERROR -- consistent with #38130/#38441, which
+            # deliberately downgraded screenshot timeout logs the same way.
+            logger.warning(
+                "Tiled screenshot time budget exhausted on tile %s/%s: "
+                "%s/%s tiles captured so far, %.1fs elapsed of a %.1fs "
+                "budget. Aborting instead of capturing remaining tiles "
+                "unchecked.%s",
+                i + 1,
+                num_tiles,
+                len(screenshot_tiles),
+                num_tiles,
+                elapsed,
+                wait_budget_seconds,
+                context_suffix,
+            )
+            raise TiledScreenshotBudgetExceededError(
+                f"Tiled screenshot budget of "
+                f"{wait_budget_seconds:.1f}s exhausted "
+                f"after {len(screenshot_tiles)}/{num_tiles} tiles"
+            )
 
         for i in range(num_tiles):
+            # Check the time budget before starting this tile's readiness wait.
+            # If it's already exhausted, we can no longer verify this (or any
+            # later) tile is actually ready to capture -- fail loudly instead
+            # of silently snapshotting a spinner or blank chart, or running
+            # past the Celery task time limit and getting SIGKILLed.
+            elapsed = time.monotonic() - start_time
+            remaining_budget = wait_budget_seconds - elapsed
+            _raise_if_budget_exhausted(elapsed, remaining_budget)
+
             # Calculate scroll position to show this tile's content
             scroll_y = dashboard_top + (i * tile_height)
 
@@ -332,17 +405,31 @@ def take_tiled_screenshot(
             )
             # Wait for scroll to settle and content to load
             page.wait_for_timeout(SCROLL_SETTLE_TIMEOUT_MS)
+
+            # Recompute the remaining budget after the scroll-settle sleep --
+            # which itself consumes real wall-clock time -- rather than
+            # reusing the value from before it, so the readiness-check
+            # timeout below is capped against a fresh number instead of a
+            # stale one that would let each tile overrun the budget by up
+            # to one settle interval.
+            tile_wait_start = time.monotonic()
+            elapsed = tile_wait_start - start_time
+            remaining_budget = wait_budget_seconds - elapsed
+            _raise_if_budget_exhausted(elapsed, remaining_budget)
+
             # Wait for every chart holder visible in the current viewport to reach
-            # a terminal state (rendered chart or error/empty state). Only check
+            # a terminal state (rendered chart or error/empty state), capped at
+            # whatever remains of the total time budget so a slow dashboard
+            # degrades gracefully instead of exceeding it. Only check
             # viewport-visible chart holders to avoid blocking on virtualization
             # placeholders rendered for off-screen charts. A holder that hasn't
             # mounted anything yet does not satisfy this check -- unlike checking
             # for the absence of `.loading`, which passes vacuously in that case.
-            tile_wait_start = time.monotonic()
+            tile_load_wait = min(load_wait, remaining_budget)
             try:
                 page.wait_for_function(
                     CHART_HOLDERS_READY_JS,
-                    timeout=load_wait * 1000,
+                    timeout=tile_load_wait * 1000,
                 )
             except PlaywrightTimeout:
                 elapsed = time.monotonic() - tile_wait_start
@@ -354,14 +441,21 @@ def take_tiled_screenshot(
                 # made the same call for the other screenshot timeout paths.
                 logger.warning(
                     "Timed out after %.2fs waiting for %s chart container(s) to "
-                    "become ready on tile %s/%s (load_wait=%ss)%s; unready chart "
-                    "holders (chart id, state): %s. Aborting tiled screenshot "
-                    "rather than capturing a blank or partially-loaded tile.",
+                    "become ready on tile %s/%s (waited %.1fs of a %ss requested "
+                    "load_wait; %.1fs elapsed of a %.1fs total budget; %s/%s "
+                    "tiles captured so far)%s; unready chart holders (chart id, "
+                    "state): %s. Aborting tiled screenshot rather than capturing "
+                    "a blank or partially-loaded tile.",
                     elapsed,
                     len(unready_chart_holders),
                     i + 1,
                     num_tiles,
+                    tile_load_wait,
                     load_wait,
+                    time.monotonic() - start_time,
+                    wait_budget_seconds,
+                    len(screenshot_tiles),
+                    num_tiles,
                     context_suffix,
                     unready_chart_holders,
                 )
@@ -377,12 +471,36 @@ def take_tiled_screenshot(
                     load_wait,
                     context_suffix,
                 )
+            readiness_wait_elapsed = time.monotonic() - tile_wait_start
 
             # Wait for chart animations (e.g. ECharts) to finish after spinner clears.
             # The global animation wait before tiling only covers the first tile;
-            # subsequent tiles need their own wait after data loads.
+            # subsequent tiles need their own wait after data loads. Capped at
+            # whatever remains of the budget; unlike the readiness wait above this
+            # is cosmetic settling, not a readiness check, so we simply skip it
+            # (rather than raise) once the budget runs out.
+            animation_wait_elapsed = 0.0
             if animation_wait > 0:
-                page.wait_for_timeout(animation_wait * 1000)
+                elapsed = time.monotonic() - start_time
+                remaining_budget = wait_budget_seconds - elapsed
+                tile_animation_wait = max(0, min(animation_wait, remaining_budget))
+                if tile_animation_wait > 0:
+                    animation_wait_start = time.monotonic()
+                    page.wait_for_timeout(tile_animation_wait * 1000)
+                    animation_wait_elapsed = time.monotonic() - animation_wait_start
+
+            # Per-tile timing breakdown so slow dashboards can be profiled from
+            # logs alone. DEBUG rather than INFO: this fires once per tile, and
+            # large dashboards can have dozens of tiles per report run.
+            logger.debug(
+                "Tile %s/%s timing: %.2fs waiting for chart readiness, "
+                "%.2fs waiting for animations.%s",
+                i + 1,
+                num_tiles,
+                readiness_wait_elapsed,
+                animation_wait_elapsed,
+                context_suffix,
+            )
 
             # Calculate what portion of the element we want to capture for this tile
             tile_start_in_element = i * tile_height
@@ -431,6 +549,12 @@ def take_tiled_screenshot(
 
         return combined_screenshot
 
+    except TiledScreenshotBudgetExceededError:
+        # Budget exhaustion must fail cleanly, not be swallowed into the
+        # generic `return None` degradation below -- the raise carries the
+        # budget diagnostics to the caller, which fails the capture loudly
+        # (#42273) instead of receiving an anonymous empty result.
+        raise
     except Exception as e:
         if readiness_timeout:
             # Let the per-tile readiness timeout propagate so the caller

superset/utils/screenshots.py

@@ -85,6 +85,26 @@ class ScreenshotCachePayloadType(TypedDict):
     status: str
 
 
+# Magic bytes for a cheap image sanity check. This is intentionally not a full
+# decode: it's meant to catch 0-byte/corrupt/blank payloads before they're
+# cached or served, not to validate the image is renderable.
+PNG_MAGIC_BYTES = b"\x89PNG\r\n\x1a\n"
+JPEG_MAGIC_BYTES = b"\xff\xd8\xff"
+
+
+def validate_screenshot_image(image: bytes | None) -> str | None:
+    """Cheaply validate screenshot bytes before they're cached or served.
+
+    :return: None if the bytes look like a usable image, otherwise a short
+        reason ("empty" or "undecodable") suitable for logging.
+    """
+    if not image:
+        return "empty"
+    if not image.startswith((PNG_MAGIC_BYTES, JPEG_MAGIC_BYTES)):
+        return "undecodable"
+    return None
+
+
 class ScreenshotCachePayload:
     def __init__(
         self,
@@ -147,6 +167,13 @@ class ScreenshotCachePayload:
     def get_status(self) -> str:
         return self.status.value
 
+    def get_invalid_image_reason(self) -> str | None:
+        """Reason this payload's image should not be served/cached, or None if
+        it passes validation (or it isn't claiming a successful screenshot)."""
+        if self.status != StatusValues.UPDATED:
+            return None
+        return validate_screenshot_image(self._image)
+
     def is_error_cache_ttl_expired(self) -> bool:
         error_cache_ttl = app.config["THUMBNAIL_ERROR_CACHE_TTL"]
         return (
@@ -263,6 +290,14 @@ class BaseScreenshot:
             elif isinstance(payload, dict):
                 payload = cast(ScreenshotCachePayloadType, payload)
                 payload = ScreenshotCachePayload.from_dict(payload)
+            if invalid_reason := payload.get_invalid_image_reason():
+                logger.warning(
+                    "Rejecting cached screenshot for %s: %s image payload; "
+                    "treating as a cache miss",
+                    cache_key,
+                    invalid_reason,
+                )
+                return None
             return payload
         logger.info("Failed at getting from cache: %s", cache_key)
         return None
@@ -331,15 +366,28 @@ class BaseScreenshot:
                         image = None
 
                 # Cache the result (success or error) to avoid immediate retries
-                if image:
+                invalid_reason = validate_screenshot_image(image)
+                # `image and` is redundant at runtime (validate_screenshot_image
+                # only returns None for truthy, well-formed bytes) but mypy can't
+                # infer that image is non-None from invalid_reason being None
+                # across the function-call boundary, so it's kept for narrowing.
+                if image and invalid_reason is None:
                     with event_logger.log_context(
                         f"screenshot.cache.{self.thumbnail_type}"
                     ):
                         cache_payload.update(image)
-                elif cache_payload.status != StatusValues.ERROR:
-                    # Only call error() if not already set — avoids overwriting
-                    # the timestamp recorded when the actual failure occurred above.
-                    cache_payload.error()
+                else:
+                    if invalid_reason:
+                        logger.warning(
+                            "Not caching screenshot result for %s: %s image payload",
+                            cache_key,
+                            invalid_reason,
+                        )
+                    if cache_payload.status != StatusValues.ERROR:
+                        # Only call error() if not already set — avoids overwriting
+                        # the timestamp recorded when the actual failure occurred
+                        # above.
+                        cache_payload.error()
 
                 logger.info("Caching thumbnail: %s", cache_key)
                 self.cache.set(cache_key, cache_payload.to_dict())

superset/utils/webdriver.py

@@ -47,6 +47,7 @@ from superset.utils.screenshot_utils import (
     CHART_HOLDERS_READY_JS,
     FIND_CHART_HOLDER_STATES_JS,
     resolve_screenshot_task_budget_seconds,
+    ScreenshotTaskBudgetExceededError,
     take_tiled_screenshot,
 )
 
@@ -61,10 +62,6 @@ PLAYWRIGHT_INSTALL_MESSAGE = (
 )
 
 
-class ScreenshotTaskBudgetExceededError(RuntimeError):
-    """Raised when no safe task budget remains before screenshot capture."""
-
-
 if TYPE_CHECKING:
     from typing import Any
 

superset/views/datasource/views.py

@@ -202,6 +202,23 @@ class Datasource(BaseSupersetView):
             payload = SamplesPayloadSchema().load(request.json)
         except ValidationError as err:
             return json_error_response(err.messages, status=400)
+
+        # Refuse early for datasource types that don't model raw rows
+        # (e.g. semantic views, which only expose pre-defined metrics and
+        # dimensions). Without this gate the request would still go through
+        # the standard query pipeline and fail with an opaque 500.
+        # ``supports_samples`` defaults to True for any datasource class that
+        # doesn't explicitly opt out, so SqlaTable/Query/SavedQuery continue
+        # to work without needing the attribute declared on each class.
+        ds_class = DatasourceDAO.sources.get(
+            DatasourceType(params["datasource_type"]),
+        )
+        if ds_class is not None and not getattr(ds_class, "supports_samples", True):
+            return json_error_response(
+                _("Samples are not available for this datasource type."),
+                status=400,
+            )
+
         dashboard_id = None
         if security_manager.is_guest_user():
             if not params["dashboard_id"]:

tests/unit_tests/common/test_query_actions_drill_detail.py

@@ -0,0 +1,70 @@
+# 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.
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from superset.common.query_actions import _get_drill_detail
+from superset.exceptions import QueryObjectValidationError
+
+
+def test_get_drill_detail_refuses_datasource_that_opts_out() -> None:
+    """
+    A datasource with ``supports_drill_to_detail = False`` (e.g. semantic
+    views) must be hard-blocked on the server. Without this gate the request
+    would fall through to ``_get_full`` and fail with an opaque error, and
+    the flag would only be enforced by the frontend menu — leaving the
+    chart-data API endpoint accepting drill-detail requests it shouldn't.
+    """
+    datasource = MagicMock()
+    datasource.supports_drill_to_detail = False
+
+    query_obj = MagicMock()
+    query_obj.datasource = datasource
+
+    query_context = MagicMock()
+
+    with pytest.raises(
+        QueryObjectValidationError,
+        match="Drill to detail is not available",
+    ):
+        _get_drill_detail(query_context, query_obj)
+
+
+def test_get_drill_detail_allows_datasource_without_flag() -> None:
+    """
+    Datasources that don't declare the flag (e.g. legacy ``SqlaTable``
+    subclasses via ``getattr`` default) must continue to work — the gate
+    only fires when the flag is explicitly ``False``.
+    """
+    datasource = MagicMock(spec=["columns"])
+    column = MagicMock()
+    column.column_name = "id"
+    datasource.columns = [column]
+
+    query_obj = MagicMock()
+    query_obj.datasource = datasource
+    query_obj.columns = []
+
+    query_context = MagicMock()
+
+    expected_payload: dict[str, list[dict[str, str]]] = {"data": []}
+    with patch(
+        "superset.common.query_actions._get_full", return_value=expected_payload
+    ) as mock_get_full:
+        assert _get_drill_detail(query_context, query_obj) is expected_payload
+    mock_get_full.assert_called_once()

tests/unit_tests/semantic_layers/models_test.py

@@ -653,6 +653,15 @@ def test_semantic_view_data(
         assert data["table_name"] == "Orders View"
         assert data["datasource_name"] == "Orders View"
         assert data["offset"] == 0
+        # Semantic views don't model raw rows, so neither samples nor
+        # drill-to-detail are available.
+        assert data["supports_samples"] is False
+        assert data["supports_drill_to_detail"] is False
+
+
+def test_semantic_view_supports_samples_is_false() -> None:
+    """The class-level flag opts SemanticView out of the Samples affordance."""
+    assert SemanticView.supports_samples is False
 
 
 @pytest.fixture
@@ -767,6 +776,11 @@ def test_semantic_view_data_populates_time_grain_sqla(
     assert grain_durations == sorted(["PT1H", "P1D", "P1M"])
 
 
+def test_semantic_view_supports_drill_to_detail_is_false() -> None:
+    """The class-level flag opts SemanticView out of Drill to detail."""
+    assert SemanticView.supports_drill_to_detail is False
+
+
 def test_semantic_view_get_query_result(
     mock_implementation: MagicMock,
 ) -> None:

tests/unit_tests/utils/screenshot_test.py

@@ -33,6 +33,10 @@ from superset.utils.screenshots import (
 
 BASE_SCREENSHOT_PATH = "superset.utils.screenshots.BaseScreenshot"
 
+# A minimal valid PNG header, used wherever a test needs bytes that pass
+# ScreenshotCachePayload's image validation.
+FAKE_PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"fake-png-body"
+
 
 class MockCache:
     """A class to manage screenshot cache."""
@@ -92,7 +96,7 @@ def test_get_cache_key(app_context, screenshot_obj):
 def test_get_from_cache_key(mocker: MockerFixture, screenshot_obj):
     """get_from_cache_key should always return a ScreenshotCachePayload Object"""
     # backwards compatibility test for retrieving plain bytes
-    fake_bytes = b"fake_screenshot_data"
+    fake_bytes = FAKE_PNG_BYTES
     BaseScreenshot.cache = MockCache()
     BaseScreenshot.cache.set("key", fake_bytes)
     cache_payload = screenshot_obj.get_from_cache_key("key")
@@ -108,10 +112,10 @@ class TestComputeAndCache:
             BASE_SCREENSHOT_PATH + ".get_from_cache_key", return_value=None
         )
         get_screenshot = mocker.patch(
-            BASE_SCREENSHOT_PATH + ".get_screenshot", return_value=b"new_image_data"
+            BASE_SCREENSHOT_PATH + ".get_screenshot", return_value=FAKE_PNG_BYTES
         )
         resize_image = mocker.patch(
-            BASE_SCREENSHOT_PATH + ".resize_image", return_value=b"resized_image_data"
+            BASE_SCREENSHOT_PATH + ".resize_image", return_value=FAKE_PNG_BYTES
         )
         BaseScreenshot.cache = MockCache()
         return {

tests/unit_tests/utils/test_screenshot_cache_fix.py

@@ -37,6 +37,10 @@ from superset.utils.screenshots import (
 BASE_SCREENSHOT_PATH = "superset.utils.screenshots.BaseScreenshot"
 DISTRIBUTED_LOCK_PATH = "superset.utils.screenshots.DistributedLock"
 
+# A minimal valid PNG header, used wherever a test needs bytes that pass
+# ScreenshotCachePayload's image validation.
+FAKE_PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"fake-png-body"
+
 
 class MockCache:
     """A class to manage screenshot cache for testing."""
@@ -83,11 +87,11 @@ class TestCacheOnlyOnSuccess:
         mocker.patch(DISTRIBUTED_LOCK_PATH)
         mocker.patch(BASE_SCREENSHOT_PATH + ".get_from_cache_key", return_value=None)
         get_screenshot = mocker.patch(
-            BASE_SCREENSHOT_PATH + ".get_screenshot", return_value=b"image_data"
+            BASE_SCREENSHOT_PATH + ".get_screenshot", return_value=FAKE_PNG_BYTES
         )
         # Mock resize_image to avoid PIL errors with fake image data
         mocker.patch(
-            BASE_SCREENSHOT_PATH + ".resize_image", return_value=b"resized_image_data"
+            BASE_SCREENSHOT_PATH + ".resize_image", return_value=FAKE_PNG_BYTES
         )
         BaseScreenshot.cache = MockCache()
         return get_screenshot
@@ -161,13 +165,15 @@ class TestCacheOnlyOnSuccess:
         screenshot_obj: BaseScreenshot,
         mock_user: MagicMock,
     ) -> None:
-        """Empty bytes from get_screenshot must set ERROR, not leave COMPUTING."""
+        """Empty bytes from get_screenshot must set ERROR, not leave COMPUTING,
+        and must log a WARNING that includes the cache key."""
         mocker.patch(DISTRIBUTED_LOCK_PATH)
         mocker.patch(BASE_SCREENSHOT_PATH + ".get_from_cache_key", return_value=None)
         mocker.patch(
             BASE_SCREENSHOT_PATH + ".get_screenshot",
             return_value=b"",
         )
+        mock_logger = mocker.patch("superset.utils.screenshots.logger")
         BaseScreenshot.cache = MockCache()
 
         screenshot_obj.compute_and_cache(user=mock_user, force=True)
@@ -177,6 +183,43 @@ class TestCacheOnlyOnSuccess:
         assert cached_value is not None
         assert cached_value["status"] == "Error"
         assert cached_value.get("image") is None
+        assert any(
+            cache_key in call.args and "empty" in call.args
+            for call in mock_logger.warning.call_args_list
+        )
+
+    def test_cache_error_status_when_screenshot_returns_garbage_bytes(
+        self,
+        mocker: MockerFixture,
+        screenshot_obj: BaseScreenshot,
+        mock_user: MagicMock,
+    ) -> None:
+        """Non-empty bytes without a valid image header must set ERROR, not be
+        cached as a success, and must log a WARNING that includes the cache key."""
+        mocker.patch(DISTRIBUTED_LOCK_PATH)
+        mocker.patch(BASE_SCREENSHOT_PATH + ".get_from_cache_key", return_value=None)
+        mocker.patch(
+            BASE_SCREENSHOT_PATH + ".get_screenshot",
+            return_value=b"this-is-not-a-real-image",
+        )
+        mocker.patch(
+            BASE_SCREENSHOT_PATH + ".resize_image",
+            return_value=b"this-is-not-a-real-image",
+        )
+        mock_logger = mocker.patch("superset.utils.screenshots.logger")
+        BaseScreenshot.cache = MockCache()
+
+        screenshot_obj.compute_and_cache(user=mock_user, force=True)
+
+        cache_key = screenshot_obj.get_cache_key()
+        cached_value = BaseScreenshot.cache.get(cache_key)
+        assert cached_value is not None
+        assert cached_value["status"] == "Error"
+        assert cached_value.get("image") is None
+        assert any(
+            cache_key in call.args and "undecodable" in call.args
+            for call in mock_logger.warning.call_args_list
+        )
 
     def test_computing_status_written_to_cache_early(
         self,
@@ -197,14 +240,14 @@ class TestCacheOnlyOnSuccess:
                 "Cache should be set to COMPUTING before screenshot starts"
             )
             assert cached_value["status"] == "Computing"
-            return b"image_data"
+            return FAKE_PNG_BYTES
 
         mocker.patch(
             BASE_SCREENSHOT_PATH + ".get_screenshot",
             side_effect=check_cache_during_screenshot,
         )
         mocker.patch(
-            BASE_SCREENSHOT_PATH + ".resize_image", return_value=b"resized_image_data"
+            BASE_SCREENSHOT_PATH + ".resize_image", return_value=FAKE_PNG_BYTES
         )
 
         screenshot_obj.compute_and_cache(user=mock_user, force=True)
@@ -429,11 +472,11 @@ class TestIntegrationCacheBugFix:
         BaseScreenshot.cache.set(cache_key, stale_payload.to_dict())
 
         mocker.patch(
-            BASE_SCREENSHOT_PATH + ".get_screenshot", return_value=b"recovered_image"
+            BASE_SCREENSHOT_PATH + ".get_screenshot", return_value=FAKE_PNG_BYTES
         )
         # Mock resize to avoid PIL errors
         mocker.patch(
-            BASE_SCREENSHOT_PATH + ".resize_image", return_value=b"resized_image"
+            BASE_SCREENSHOT_PATH + ".resize_image", return_value=FAKE_PNG_BYTES
         )
 
         # Should trigger task because COMPUTING is stale
@@ -482,3 +525,72 @@ class TestIntegrationCacheBugFix:
 
         assert payload._image == old_image
         assert payload.status == StatusValues.COMPUTING
+
+
+class TestReadSideImageValidation:
+    """A cached payload that claims a successful screenshot (status UPDATED)
+    but carries invalid image bytes must be served as a cache miss, not
+    returned to the caller — this is what the dashboard/chart screenshot
+    endpoints call to fetch bytes to serve."""
+
+    def test_zero_byte_image_is_treated_as_cache_miss(
+        self, mocker: MockerFixture, screenshot_obj: BaseScreenshot
+    ) -> None:
+        mock_logger = mocker.patch("superset.utils.screenshots.logger")
+        BaseScreenshot.cache = MockCache()
+        cache_key = screenshot_obj.get_cache_key()
+        stale_payload = ScreenshotCachePayload(image=b"", status=StatusValues.UPDATED)
+        BaseScreenshot.cache.set(cache_key, stale_payload.to_dict())
+
+        result = screenshot_obj.get_from_cache_key(cache_key)
+
+        assert result is None
+        assert any(
+            cache_key in call.args and "empty" in call.args
+            for call in mock_logger.warning.call_args_list
+        )
+
+    def test_garbage_bytes_image_is_treated_as_cache_miss(
+        self, mocker: MockerFixture, screenshot_obj: BaseScreenshot
+    ) -> None:
+        mock_logger = mocker.patch("superset.utils.screenshots.logger")
+        BaseScreenshot.cache = MockCache()
+        cache_key = screenshot_obj.get_cache_key()
+        garbage_payload = ScreenshotCachePayload(image=b"not-an-image-at-all")
+        BaseScreenshot.cache.set(cache_key, garbage_payload.to_dict())
+
+        result = screenshot_obj.get_from_cache_key(cache_key)
+
+        assert result is None
+        assert any(
+            cache_key in call.args and "undecodable" in call.args
+            for call in mock_logger.warning.call_args_list
+        )
+
+    def test_valid_image_is_served_normally(
+        self, screenshot_obj: BaseScreenshot
+    ) -> None:
+        BaseScreenshot.cache = MockCache()
+        cache_key = screenshot_obj.get_cache_key()
+        valid_payload = ScreenshotCachePayload(image=FAKE_PNG_BYTES)
+        BaseScreenshot.cache.set(cache_key, valid_payload.to_dict())
+
+        result = screenshot_obj.get_from_cache_key(cache_key)
+
+        assert result is not None
+        assert result.get_image().read() == FAKE_PNG_BYTES
+
+    def test_pending_status_with_no_image_is_not_rejected(
+        self, screenshot_obj: BaseScreenshot
+    ) -> None:
+        """Non-UPDATED statuses (e.g. PENDING/COMPUTING) aren't claiming a
+        successful screenshot, so they should be returned as-is."""
+        BaseScreenshot.cache = MockCache()
+        cache_key = screenshot_obj.get_cache_key()
+        pending_payload = ScreenshotCachePayload(status=StatusValues.PENDING)
+        BaseScreenshot.cache.set(cache_key, pending_payload.to_dict())
+
+        result = screenshot_obj.get_from_cache_key(cache_key)
+
+        assert result is not None
+        assert result.status == StatusValues.PENDING

tests/unit_tests/utils/test_screenshot_utils.py

@@ -25,8 +25,11 @@ from superset.utils.screenshot_utils import (
     combine_screenshot_tiles,
     resolve_screenshot_task_budget_seconds,
     SCREENSHOT_TASK_BUDGET_MAX_MARGIN_SECONDS,
+    ScreenshotTaskBudgetExceededError,
     SCROLL_SETTLE_TIMEOUT_MS,
     take_tiled_screenshot,
+    TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS,
+    TiledScreenshotBudgetExceededError,
 )
 
 
@@ -453,12 +456,17 @@ class TestTakeTiledScreenshot:
         assert warning_args[2] == 1  # count of unready chart containers
         assert warning_args[3] == 1  # tile index
         assert warning_args[4] == 3  # total tiles
-        assert warning_args[5] == 30  # load_wait
-        assert warning_args[6] == ""  # no log_context passed
+        assert warning_args[5] == 30  # tile_load_wait (uncapped: budget remains)
+        assert warning_args[6] == 30  # requested load_wait
+        assert isinstance(warning_args[7], float)  # total elapsed vs budget
+        assert warning_args[8] == 1440  # total budget (fixed fallback)
+        assert warning_args[9] == 0  # tiles captured so far
+        assert warning_args[10] == 3  # total tiles
+        assert warning_args[11] == ""  # no log_context passed
         # Diagnostic payload identifies chart id AND the state it's stuck in
         # (spinner mounted vs nothing mounted vs waiting-on-database) so a
         # slow query can be told apart from the virtualization race.
-        assert warning_args[7] == [{"chartId": "42", "state": "waiting_on_database"}]
+        assert warning_args[12] == [{"chartId": "42", "state": "waiting_on_database"}]
 
     def test_timeout_warning_includes_log_context(self, mock_page):
         """The log context (e.g. report execution id) is threaded through for
@@ -484,7 +492,7 @@ class TestTakeTiledScreenshot:
                     )
 
         warning_args = mock_logger.warning.call_args[0]
-        assert warning_args[6] == " [execution_id=abc-123]"
+        assert warning_args[11] == " [execution_id=abc-123]"
 
     def test_chart_holder_with_nothing_mounted_blocks_wait(self, mock_page):
         """Regression test for the vacuous-pass race (PR #39895).
@@ -646,3 +654,311 @@ class TestTakeTiledScreenshot:
 
         sig = inspect.signature(take_tiled_screenshot)
         assert sig.parameters["animation_wait"].default == 0
+
+
+class TestTileWaitBudget:
+    """The tiled operation's cumulative per-tile waits are capped by one
+    wall-clock budget derived from the running Celery task's own time limit
+    (resolve_screenshot_task_budget_seconds), falling back to a fixed total
+    ceiling outside Celery because per-tile waits accumulate."""
+
+    @pytest.fixture
+    def mock_page(self):
+        """Create a mock Playwright page object for a 3-tile (5000px) dashboard."""
+        page = MagicMock()
+        element = MagicMock()
+        page.locator.return_value = element
+        page.evaluate.return_value = {
+            "height": 5000,
+            "top": 100,
+            "left": 50,
+            "width": 800,
+        }
+        page.screenshot.return_value = b"fake_screenshot_data"
+        return page
+
+    class _FakeClock:
+        """Stateful monotonic() stand-in the test advances explicitly.
+
+        Robust to how many times the code under test samples the clock per
+        tile (budget check, per-tile wait timing, animation budget) -- only
+        explicit advances move time forward.
+        """
+
+        def __init__(self) -> None:
+            self.now = 0.0
+
+        def __call__(self) -> float:
+            return self.now
+
+    def test_budget_error_is_task_budget_error_subclass(self):
+        """Callers can catch the whole budget-error family with the base
+        ScreenshotTaskBudgetExceededError type."""
+        assert issubclass(
+            TiledScreenshotBudgetExceededError, ScreenshotTaskBudgetExceededError
+        )
+
+    def test_per_tile_wait_shrinks_as_budget_depletes(self, mock_page, monkeypatch):
+        """Each tile's readiness-wait timeout is capped at the remaining budget."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        clock = self._FakeClock()
+        # Simulate slow tiles: the readiness wait itself consumes wall time,
+        # so each subsequent tile sees less remaining budget.
+        wait_durations = iter([950, 40, 5])
+
+        def slow_wait(*args, **kwargs):
+            clock.now += next(wait_durations)
+
+        mock_page.wait_for_function.side_effect = slow_wait
+
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    result = take_tiled_screenshot(
+                        mock_page, "dashboard", tile_height=2000, load_wait=100
+                    )
+
+        assert result is not None
+        timeouts = [
+            call[1]["timeout"] for call in mock_page.wait_for_function.call_args_list
+        ]
+        # remaining budget at each tile's wait: 1000, 50, 10 seconds
+        # -> capped timeouts shrink
+        assert timeouts == [100 * 1000, 50 * 1000, 10 * 1000]
+        assert timeouts == sorted(timeouts, reverse=True)
+
+    def test_readiness_wait_uses_budget_recomputed_after_scroll_settle(
+        self, mock_page, monkeypatch
+    ):
+        """The readiness-wait timeout must be capped using the budget
+        recomputed *after* the scroll-settle sleep, not the stale value from
+        before it -- otherwise each tile could overrun the total budget by up
+        to one settle interval."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        # A single-tile dashboard to keep the scenario simple.
+        mock_page.evaluate.return_value = {
+            "height": 1000,
+            "top": 100,
+            "left": 50,
+            "width": 800,
+        }
+        clock = self._FakeClock()
+        # The scroll-settle sleep itself consumes 950s of wall-clock time,
+        # leaving only 50s of the 1000s budget by the time the readiness
+        # wait is capped.
+        mock_page.wait_for_timeout.side_effect = lambda *args, **kwargs: setattr(
+            clock, "now", clock.now + 950
+        )
+
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    take_tiled_screenshot(
+                        mock_page, "dashboard", tile_height=2000, load_wait=999
+                    )
+
+        timeout = mock_page.wait_for_function.call_args_list[0][1]["timeout"]
+        # Must reflect the post-settle remaining budget (50s), not the
+        # stale pre-settle value (1000s, which would have let load_wait's
+        # full 999s through uncapped).
+        assert timeout == 50 * 1000
+
+    def test_budget_exhausted_raises_and_stops_capturing(self, mock_page, monkeypatch):
+        """Exhausting the budget aborts cleanly instead of capturing unchecked."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        clock = self._FakeClock()
+        # Tile 0's readiness wait consumes the whole budget; tile 1's budget
+        # check then sees remaining <= 0 and raises before capturing.
+        mock_page.wait_for_function.side_effect = lambda *args, **kwargs: setattr(
+            clock, "now", 1000.0
+        )
+
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch(
+                    "superset.utils.screenshot_utils.combine_screenshot_tiles"
+                ) as mock_combine:
+                    with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                        with pytest.raises(TiledScreenshotBudgetExceededError):
+                            take_tiled_screenshot(
+                                mock_page, "dashboard", tile_height=2000, load_wait=100
+                            )
+
+        # Only the first tile was captured before the budget ran out.
+        assert mock_page.screenshot.call_count == 1
+        # Tiles were never combined -- the function raised before that point.
+        mock_combine.assert_not_called()
+
+        # Budget exhaustion is a customer chart-loading issue, not a Superset
+        # system fault, so it must log at WARNING (not ERROR) -- consistent
+        # with the #38130/#38441 precedent for screenshot timeout logging.
+        assert mock_logger.error.call_count == 0
+        mock_logger.warning.assert_called_once()
+        warning_args = mock_logger.warning.call_args[0]
+        assert "budget exhausted" in warning_args[0]
+        # tile index, tiles total, tiles captured, tiles total,
+        # elapsed seconds, budget seconds, log-context suffix
+        assert warning_args[1] == 2
+        assert warning_args[2] == 3
+        assert warning_args[3] == 1
+        assert warning_args[4] == 3
+        assert warning_args[5] == 1000
+        assert warning_args[6] == 1000
+        assert warning_args[7] == ""
+
+    def test_budget_exhausted_warning_includes_log_context(
+        self, mock_page, monkeypatch
+    ):
+        """log_context (e.g. report execution id) is appended to the warning."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        clock = self._FakeClock()
+        # Tile 0's readiness wait consumes the whole budget; tile 1's budget
+        # check then sees remaining <= 0 and raises.
+        mock_page.wait_for_function.side_effect = lambda *args, **kwargs: setattr(
+            clock, "now", 1000.0
+        )
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                        with pytest.raises(TiledScreenshotBudgetExceededError):
+                            take_tiled_screenshot(
+                                mock_page,
+                                "dashboard",
+                                tile_height=2000,
+                                load_wait=100,
+                                log_context="execution_id=abc-123",
+                            )
+
+        warning_args = mock_logger.warning.call_args[0]
+        assert warning_args[-1] == " [execution_id=abc-123]"
+
+    def test_budget_exhausted_before_first_tile_raises_without_capture(
+        self, mock_page, monkeypatch
+    ):
+        """No budget floor: a budget already exhausted by setup (element
+        lookup/dimension probing) raises before the first tile is captured,
+        matching the non-tiled path's raise-before-capture semantics."""
+        monkeypatch.setattr(
+            "superset.utils.screenshot_utils.TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS",  # noqa: E501
+            1000,
+        )
+        clock = self._FakeClock()
+        # The dashboard-dimension evaluate() itself consumes the whole budget.
+        original_return = {"height": 5000, "top": 100, "left": 50, "width": 800}
+
+        def slow_evaluate(*args, **kwargs):
+            clock.now = 1000.0
+            return original_return
+
+        mock_page.evaluate.side_effect = slow_evaluate
+
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch(
+                    "superset.utils.screenshot_utils.combine_screenshot_tiles"
+                ) as mock_combine:
+                    with pytest.raises(TiledScreenshotBudgetExceededError):
+                        take_tiled_screenshot(
+                            mock_page, "dashboard", tile_height=2000, load_wait=100
+                        )
+
+        mock_page.screenshot.assert_not_called()
+        mock_combine.assert_not_called()
+
+    def test_no_celery_context_uses_fixed_total_fallback(self, mock_page):
+        """Outside Celery the helper returns None; the tiled path must fall
+        back to the fixed total ceiling rather than running uncapped, because
+        per-tile waits accumulate across tiles."""
+        clock = self._FakeClock()
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    take_tiled_screenshot(
+                        mock_page,
+                        "dashboard",
+                        tile_height=2000,
+                        load_wait=10_000,  # deliberately above the fallback
+                    )
+
+        first_timeout = mock_page.wait_for_function.call_args_list[0][1]["timeout"]
+        assert first_timeout == TILED_SCREENSHOT_TOTAL_WAIT_BUDGET_SECONDS * 1000
+
+    def test_derived_task_budget_caps_tile_wait(self, mock_page):
+        """Inside Celery, the tiled path caps waits using the same
+        task-derived budget as the non-tiled path (helper reuse, #42427)."""
+        task = MagicMock()
+        task.request.timelimit = (120, None)  # (hard, soft): 120s hard limit
+
+        clock = self._FakeClock()
+        with patch("superset.utils.screenshot_utils.current_task", task):
+            with patch("superset.utils.screenshot_utils.time.monotonic", new=clock):
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    take_tiled_screenshot(
+                        mock_page, "dashboard", tile_height=2000, load_wait=200
+                    )
+
+        # margin = min(300, 120 * 0.2) = 24; budget = 120 - 24 = 96
+        first_timeout = mock_page.wait_for_function.call_args_list[0][1]["timeout"]
+        assert first_timeout == 96 * 1000
+        assert first_timeout < 200 * 1000
+
+    def test_fast_dashboard_matches_default_behavior(self, mock_page):
+        """Well under budget, waits are not capped and behavior is unchanged."""
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                result = take_tiled_screenshot(
+                    mock_page,
+                    "dashboard",
+                    tile_height=2000,
+                    load_wait=30,
+                    animation_wait=5,
+                )
+
+        assert result is not None
+        assert mock_page.screenshot.call_count == 3
+
+        for call in mock_page.wait_for_function.call_args_list:
+            assert call[1]["timeout"] == 30 * 1000
+
+        animation_calls = [
+            call
+            for call in mock_page.wait_for_timeout.call_args_list
+            if call[0][0] == 5 * 1000
+        ]
+        assert len(animation_calls) == 3
+
+    def test_per_tile_timing_debug_line_logged(self, mock_page):
+        """Each tile logs a DEBUG timing breakdown (readiness wait, animation
+        wait) so slow dashboards can be profiled from logs alone."""
+        with patch("superset.utils.screenshot_utils.current_task", None):
+            with patch("superset.utils.screenshot_utils.logger") as mock_logger:
+                with patch("superset.utils.screenshot_utils.combine_screenshot_tiles"):
+                    take_tiled_screenshot(
+                        mock_page,
+                        "dashboard",
+                        tile_height=2000,
+                        log_context="cache_key=xyz",
+                    )
+
+        timing_calls = [
+            call for call in mock_logger.debug.call_args_list if "timing" in call[0][0]
+        ]
+        assert len(timing_calls) == 3
+        for i, call in enumerate(timing_calls):
+            args = call[0]
+            assert args[1] == i + 1  # tile index
+            assert args[2] == 3  # total tiles
+            assert args[-1] == " [cache_key=xyz]"

tests/unit_tests/views/datasource/views_test.py

@@ -310,3 +310,66 @@ def test_save_non_editor_with_editors_field_is_rejected(
             raw_save(_view_self())
 
     mock_security_manager.raise_for_editorship.assert_called_once_with(mock_orm)
+
+
+# ---------------------------------------------------------------------------
+# Datasource.samples
+# ---------------------------------------------------------------------------
+
+
+@patch("superset.views.datasource.views._", lambda s: s)
+@patch("superset.views.datasource.views.get_samples")
+@patch("superset.views.datasource.views.json_error_response")
+@patch("superset.views.datasource.views.security_manager", new_callable=MagicMock)
+def test_samples_returns_400_for_unsupported_datasource_type(
+    mock_security_manager: MagicMock,
+    mock_json_error_response: MagicMock,
+    mock_get_samples: MagicMock,
+) -> None:
+    """Semantic views can't return raw samples — endpoint should refuse with 400."""
+    from flask import Flask
+
+    mock_security_manager.is_guest_user.return_value = False
+    mock_json_error_response.return_value = "error-response"
+
+    raw_samples = _get_view_func("samples")
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/datasource/samples?datasource_type=semantic_view&datasource_id=1",
+        method="POST",
+        json={},
+    ):
+        result = raw_samples(_view_self())
+
+    assert result == "error-response"
+    mock_json_error_response.assert_called_once()
+    _, kwargs = mock_json_error_response.call_args
+    assert kwargs.get("status") == 400
+    # The bail-out must happen before any sample fetching is attempted.
+    mock_get_samples.assert_not_called()
+
+
+@patch("superset.views.datasource.views.get_samples")
+@patch("superset.views.datasource.views.security_manager", new_callable=MagicMock)
+def test_samples_proceeds_for_supported_datasource_type(
+    mock_security_manager: MagicMock,
+    mock_get_samples: MagicMock,
+) -> None:
+    """A `query` datasource (supports_samples=True) bypasses the 400 short-circuit."""
+    from flask import Flask
+
+    mock_security_manager.is_guest_user.return_value = False
+    mock_get_samples.return_value = {"rows": []}
+
+    view = _view_self()
+    raw_samples = _get_view_func("samples")
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/datasource/samples?datasource_type=query&datasource_id=1",
+        method="POST",
+        json={},
+    ):
+        raw_samples(view)
+
+    mock_get_samples.assert_called_once()
+    view.json_response.assert_called_once_with({"result": {"rows": []}})