fix: catch no table error (#32640)

superset/db_engine_specs/presto.py

@@ -38,12 +38,14 @@ from sqlalchemy.engine.base import Engine
 from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.result import Row as ResultRow
 from sqlalchemy.engine.url import URL
+from sqlalchemy.exc import NoSuchTableError
 from sqlalchemy.sql.expression import ColumnClause, Select
 
 from superset import cache_manager, db, is_feature_enabled
 from superset.common.db_query_status import QueryStatus
 from superset.constants import TimeGrain
 from superset.db_engine_specs.base import BaseEngineSpec
+from superset.db_engine_specs.exceptions import SupersetDBAPIProgrammingError
 from superset.errors import SupersetErrorType
 from superset.exceptions import SupersetTemplateException
 from superset.models.sql_lab import Query
@@ -1250,26 +1252,31 @@ class PrestoEngineSpec(PrestoBaseEngineSpec):
     ) -> dict[str, Any]:
         metadata = {}
 
-        if indexes := database.get_indexes(table):
-            col_names, latest_parts = cls.latest_partition(
-                database,
-                table,
-                show_first=True,
-                indexes=indexes,
-            )
-
-            if not latest_parts:
-                latest_parts = tuple([None] * len(col_names))
-
-            metadata["partitions"] = {
-                "cols": sorted(indexes[0].get("column_names", [])),
-                "latest": dict(zip(col_names, latest_parts, strict=False)),
-                "partitionQuery": cls._partition_query(
-                    table=table,
+        try:
+            if indexes := database.get_indexes(table):
+                col_names, latest_parts = cls.latest_partition(
+                    database,
+                    table,
+                    show_first=True,
                     indexes=indexes,
-                    database=database,
-                ),
-            }
+                )
+
+                if not latest_parts:
+                    latest_parts = tuple([None] * len(col_names))
+
+                metadata["partitions"] = {
+                    "cols": sorted(indexes[0].get("column_names", [])),
+                    "latest": dict(zip(col_names, latest_parts, strict=False)),
+                    "partitionQuery": cls._partition_query(
+                        table=table,
+                        indexes=indexes,
+                        database=database,
+                    ),
+                }
+        except NoSuchTableError as ex:
+            raise SupersetDBAPIProgrammingError(
+                "Table doesn't seem to exist on the database"
+            ) from ex
 
         metadata["view"] = cast(
             Any,

tests/integration_tests/db_engine_specs/presto_tests.py

@@ -19,8 +19,10 @@ from textwrap import dedent
 from unittest import mock, skipUnless
 
 import pandas as pd
+import pytest
 from flask.ctx import AppContext
 from sqlalchemy import types  # noqa: F401
+from sqlalchemy.exc import NoSuchTableError
 from sqlalchemy.sql import select
 
 from superset.db_engine_specs.presto import PrestoEngineSpec
@@ -570,6 +572,27 @@ class TestPrestoDbEngineSpec(SupersetTestCase):
         assert result["partitions"]["cols"] == ["ds", "hour"]
         assert result["partitions"]["latest"] == {"ds": "01-01-19", "hour": 1}
 
+    def test_get_extra_table_metadata_no_table_found(self):
+        """
+        Test get_extra_table_metadata when a NoSuchTableError (simulating NoTableFound)
+        is raised by the database.get_df method.
+        """
+        # Setup a fake database
+        database = mock.MagicMock()
+        database.get_indexes.return_value = [
+            {"column_names": ["ds"]}
+        ]  # Return indexes so get_df is called
+        database.get_extra.return_value = {}
+        # Simulate that the table is not found
+        database.get_df.side_effect = NoSuchTableError("Table not found")
+
+        from superset.db_engine_specs.exceptions import SupersetDBAPIProgrammingError
+
+        with pytest.raises(SupersetDBAPIProgrammingError):
+            PrestoEngineSpec.get_extra_table_metadata(
+                database, Table("test_table", "test_schema")
+            )
+
     def test_presto_where_latest_partition(self):
         db = mock.Mock()
         db.get_indexes = mock.Mock(return_value=[{"column_names": ["ds", "hour"]}])