feat(SIP-95): new endpoint for table metadata

superset-frontend/src/SqlLab/actions/sqlLab.test.js

@@ -508,10 +508,10 @@ describe('async actions', () => {
     fetchMock.delete(updateTableSchemaEndpoint, {});
     fetchMock.post(updateTableSchemaEndpoint, JSON.stringify({ id: 1 }));
 
-    const getTableMetadataEndpoint = 'glob:**/api/v1/database/*/table/*/*/';
+    const getTableMetadataEndpoint = 'glob:**/api/v1/database/*/table_metadata/*';
     fetchMock.get(getTableMetadataEndpoint, {});
     const getExtraTableMetadataEndpoint =
-      'glob:**/api/v1/database/*/table_metadata/extra/';
+      'glob:**/api/v1/database/*/table_metadata/extra/*';
     fetchMock.get(getExtraTableMetadataEndpoint, {});
 
     let isFeatureEnabledMock;

superset-frontend/src/SqlLab/components/SqlEditorLeftBar/SqlEditorLeftBar.test.tsx

@@ -61,13 +61,13 @@ beforeEach(() => {
       },
     ],
   });
-  fetchMock.get('glob:*/api/v1/database/*/table/*/*', {
+  fetchMock.get('glob:*/api/v1/database/*/table_metadata/*', {
     status: 200,
     body: {
       columns: table.columns,
     },
   });
-  fetchMock.get('glob:*/api/v1/database/*/table_metadata/extra/', {
+  fetchMock.get('glob:*/api/v1/database/*/table_metadata/extra/*', {
     status: 200,
     body: {},
   });

superset-frontend/src/SqlLab/components/TableElement/TableElement.test.tsx

@@ -47,7 +47,7 @@ jest.mock(
       <div data-test="mock-column-element">{column.name}</div>
     ),
 );
-const getTableMetadataEndpoint = 'glob:**/api/v1/database/*/table/*/*/';
+const getTableMetadataEndpoint = 'glob:**/api/v1/database/*/table_metadata/*';
 const getExtraTableMetadataEndpoint =
   'glob:**/api/v1/database/*/table_metadata/extra/*';
 const updateTableSchemaEndpoint = 'glob:*/tableschemaview/*/expanded';

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

@@ -74,7 +74,9 @@ const DatasetPanelWrapper = ({
     const { dbId, tableName, schema } = props;
     setLoading(true);
     setHasColumns?.(false);
-    const path = `/api/v1/database/${dbId}/table/${tableName}/${schema}/`;
+    const path = schema
+      ? `/api/v1/database/${dbId}/table_metadata/?name=${tableName}&schema=${schema}/`
+      : `/api/v1/database/${dbId}/table_metadata/?name=${tableName}`;
     try {
       const response = await SupersetClient.get({
         endpoint: path,

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

@@ -114,9 +114,9 @@ const tableApi = api.injectEndpoints({
     }),
     tableMetadata: builder.query<TableMetaData, FetchTableMetadataQueryParams>({
       query: ({ dbId, schema, table }) => ({
-        endpoint: `/api/v1/database/${dbId}/table/${encodeURIComponent(
-          table,
-        )}/${encodeURIComponent(schema)}/`,
+        endpoint: schema
+          ? `/api/v1/database/${dbId}/table_metadata/?name=${table}&schema=${schema}`
+          : `/api/v1/database/${dbId}/table_metadata/?name=${table}`,
         transformResponse: ({ json }: TableMetadataReponse) => json,
       }),
     }),

superset/commands/dataset/create.py

@@ -34,6 +34,7 @@
 from superset.daos.exceptions import DAOCreateFailedError
 from superset.exceptions import SupersetSecurityException
 from superset.extensions import db, security_manager
+from superset.sql_parse import Table
 
 logger = logging.getLogger(__name__)
 
@@ -80,7 +81,10 @@ def validate(self) -> None:
         if (
             database
             and not sql
-            and not DatasetDAO.validate_table_exists(database, table_name, schema)
+            and not DatasetDAO.validate_table_exists(
+                database,
+                Table(table_name, schema),
+            )
         ):
             exceptions.append(TableNotFoundValidationError(table_name))
 

superset/connectors/sqla/models.py

@@ -49,7 +49,7 @@
     Integer,
     or_,
     String,
-    Table,
+    Table as DBTable,
     Text,
     update,
 )
@@ -108,7 +108,7 @@
     validate_adhoc_subquery,
 )
 from superset.models.slice import Slice
-from superset.sql_parse import ParsedQuery, sanitize_clause
+from superset.sql_parse import ParsedQuery, sanitize_clause, Table
 from superset.superset_typing import (
     AdhocColumn,
     AdhocMetric,
@@ -1068,7 +1068,7 @@ def data(self) -> dict[str, Any]:
         return {s: getattr(self, s) for s in attrs}
 
 
-sqlatable_user = Table(
+sqlatable_user = DBTable(
     "sqlatable_user",
     metadata,
     Column("id", Integer, primary_key=True),
@@ -1146,6 +1146,7 @@ class SqlaTable(
         foreign_keys=[database_id],
     )
     schema = Column(String(255))
+    catalog = Column(String(256), nullable=True, default=None)
     sql = Column(MediumText())
     is_sqllab_view = Column(Boolean, default=False)
     template_params = Column(Text)
@@ -1322,8 +1323,7 @@ def external_metadata(self) -> list[ResultSetColumnType]:
             return get_virtual_table_metadata(dataset=self)
         return get_physical_table_metadata(
             database=self.database,
-            table_name=self.table_name,
-            schema_name=self.schema,
+            table=Table(self.table_name, self.schema, self.catalog),
             normalize_columns=self.normalize_columns,
         )
 
@@ -1339,7 +1339,9 @@ def select_star(self) -> str | None:
         # show_cols and latest_partition set to false to avoid
         # the expensive cost of inspecting the DB
         return self.database.select_star(
-            self.table_name, schema=self.schema, show_cols=False, latest_partition=False
+            Table(self.table_name, self.schema, self.catalog),
+            show_cols=False,
+            latest_partition=False,
         )
 
     @property
@@ -1779,7 +1781,13 @@ def assign_column_label(df: pd.DataFrame) -> pd.DataFrame | None:
         )
 
     def get_sqla_table_object(self) -> Table:
-        return self.database.get_table(self.table_name, schema=self.schema)
+        return self.database.get_table(
+            Table(
+                self.table_name,
+                self.schema,
+                self.catalog,
+            )
+        )
 
     def fetch_metadata(self, commit: bool = True) -> MetadataResult:
         """
@@ -1791,7 +1799,13 @@ def fetch_metadata(self, commit: bool = True) -> MetadataResult:
         new_columns = self.external_metadata()
         metrics = [
             SqlMetric(**metric)
-            for metric in self.database.get_metrics(self.table_name, self.schema)
+            for metric in self.database.get_metrics(
+                Table(
+                    self.table_name,
+                    self.schema,
+                    self.catalog,
+                )
+            )
         ]
         any_date_col = None
         db_engine_spec = self.db_engine_spec
@@ -2038,15 +2052,15 @@ def load_database(self: SqlaTable) -> None:
 sa.event.listen(SqlMetric, "after_update", SqlaTable.update_column)
 sa.event.listen(TableColumn, "after_update", SqlaTable.update_column)
 
-RLSFilterRoles = Table(
+RLSFilterRoles = DBTable(
     "rls_filter_roles",
     metadata,
     Column("id", Integer, primary_key=True),
     Column("role_id", Integer, ForeignKey("ab_role.id"), nullable=False),
     Column("rls_filter_id", Integer, ForeignKey("row_level_security_filters.id")),
 )
 
-RLSFilterTables = Table(
+RLSFilterTables = DBTable(
     "rls_filter_tables",
     metadata,
     Column("id", Integer, primary_key=True),

superset/connectors/sqla/utils.py

@@ -38,7 +38,7 @@
 )
 from superset.models.core import Database
 from superset.result_set import SupersetResultSet
-from superset.sql_parse import ParsedQuery
+from superset.sql_parse import ParsedQuery, Table
 from superset.superset_typing import ResultSetColumnType
 
 if TYPE_CHECKING:
@@ -47,24 +47,18 @@
 
 def get_physical_table_metadata(
     database: Database,
-    table_name: str,
+    table: Table,
     normalize_columns: bool,
-    schema_name: str | None = None,
 ) -> list[ResultSetColumnType]:
     """Use SQLAlchemy inspector to get table metadata"""
     db_engine_spec = database.db_engine_spec
     db_dialect = database.get_dialect()
-    # ensure empty schema
-    _schema_name = schema_name if schema_name else None
-    # Table does not exist or is not visible to a connection.
 
-    if not (
-        database.has_table_by_name(table_name=table_name, schema=_schema_name)
-        or database.has_view_by_name(view_name=table_name, schema=_schema_name)
-    ):
-        raise NoSuchTableError
+    # Table does not exist or is not visible to a connection.
+    if not (database.has_table(table) or database.has_view(table)):
+        raise NoSuchTableError(table)
 
-    cols = database.get_columns(table_name, schema=_schema_name)
+    cols = database.get_columns(table)
     for col in cols:
         try:
             if isinstance(col["type"], TypeEngine):

superset/daos/dataset.py

@@ -30,6 +30,7 @@
 from superset.models.core import Database
 from superset.models.dashboard import Dashboard
 from superset.models.slice import Slice
+from superset.sql_parse import Table
 from superset.utils.core import DatasourceType
 from superset.views.base import DatasourceFilter
 
@@ -72,13 +73,14 @@ def get_related_objects(database_id: int) -> dict[str, Any]:
 
     @staticmethod
     def validate_table_exists(
-        database: Database, table_name: str, schema: str | None
+        database: Database,
+        table: Table,
     ) -> bool:
         try:
-            database.get_table(table_name, schema=schema)
+            database.get_table(table)
             return True
         except SQLAlchemyError as ex:  # pragma: no cover
-            logger.warning("Got an error %s validating table: %s", str(ex), table_name)
+            logger.warning("Got an error %s validating table: %s", str(ex), table)
             return False
 
     @staticmethod

superset/databases/api.py

@@ -133,6 +133,7 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
         RouteMethod.RELATED,
         "tables",
         "table_metadata",
+        "table_metadata_deprecated",
         "table_extra_metadata",
         "table_extra_metadata_deprecated",
         "select_star",
@@ -717,10 +718,10 @@ def tables(self, pk: int, **kwargs: Any) -> FlaskResponse:
     @statsd_metrics
     @event_logger.log_this_with_context(
         action=lambda self, *args, **kwargs: f"{self.__class__.__name__}"
-        f".table_metadata",
+        f".table_metadata_deprecated",
         log_to_statsd=False,
     )
-    def table_metadata(
+    def table_metadata_deprecated(
         self, database: Database, table_name: str, schema_name: str
     ) -> FlaskResponse:
         """Get database table metadata.
@@ -761,16 +762,16 @@ def table_metadata(
             500:
               $ref: '#/components/responses/500'
         """
-        self.incr_stats("init", self.table_metadata.__name__)
+        self.incr_stats("init", self.table_metadata_deprecated.__name__)
         try:
-            table_info = get_table_metadata(database, table_name, schema_name)
+            table_info = get_table_metadata(database, Table(table_name, schema_name))
         except SQLAlchemyError as ex:
-            self.incr_stats("error", self.table_metadata.__name__)
+            self.incr_stats("error", self.table_metadata_deprecated.__name__)
             return self.response_422(error_msg_from_exception(ex))
         except SupersetException as ex:
             return self.response(ex.status, message=ex.message)
 
-        self.incr_stats("success", self.table_metadata.__name__)
+        self.incr_stats("success", self.table_metadata_deprecated.__name__)
         return self.response(200, **table_info)
 
     @expose("/<int:pk>/table_extra/<path:table_name>/<schema_name>/", methods=("GET",))
@@ -839,7 +840,86 @@ def table_extra_metadata_deprecated(
         payload = database.db_engine_spec.get_extra_table_metadata(database, table)
         return self.response(200, **payload)
 
-    @expose("/<int:pk>/table_metadata/extra/", methods=("GET",))
+    @expose("/<int:pk>/table_metadata/", methods=["GET"])
+    @protect()
+    @statsd_metrics
+    @event_logger.log_this_with_context(
+        action=lambda self, *args, **kwargs: f"{self.__class__.__name__}"
+        f".table_metadata",
+        log_to_statsd=False,
+    )
+    def table_metadata(self, pk: int) -> FlaskResponse:
+        """
+        Get metadata for a given table.
+
+        Optionally, a schema and a catalog can be passed, if different from the default
+        ones.
+        ---
+        get:
+          summary: Get table metadata
+          description: >-
+            Metadata associated with the table (columns, indexes, etc.)
+          parameters:
+          - in: path
+            schema:
+              type: integer
+            name: pk
+            description: The database id
+          - in: query
+            schema:
+              type: string
+            name: table
+            required: true
+            description: Table name
+          - in: query
+            schema:
+              type: string
+            name: schema
+            description: >-
+              Optional table schema, if not passed default schema will be used
+          - in: query
+            schema:
+              type: string
+            name: catalog
+            description: >-
+              Optional table catalog, if not passed default catalog will be used
+          responses:
+            200:
+              description: Table metadata information
+              content:
+                application/json:
+                  schema:
+                    $ref: "#/components/schemas/TableExtraMetadataResponseSchema"
+            401:
+              $ref: '#/components/responses/401'
+            404:
+              $ref: '#/components/responses/404'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        self.incr_stats("init", self.table_metadata.__name__)
+
+        database = DatabaseDAO.find_by_id(pk)
+        if database is None:
+            raise DatabaseNotFoundException("No such database")
+
+        try:
+            parameters = QualifiedTableSchema().load(request.args)
+        except ValidationError as ex:
+            raise InvalidPayloadSchemaError(ex) from ex
+
+        table = Table(parameters["name"], parameters["schema"], parameters["catalog"])
+        try:
+            security_manager.raise_for_access(database=database, table=table)
+        except SupersetSecurityException as ex:
+            # instead of raising 403, raise 404 to hide table existence
+            raise TableNotFoundException("No such table") from ex
+
+        payload = database.db_engine_spec.get_table_metadata(database, table)
+
+        return self.response(200, **payload)
+
+    @expose("/<int:pk>/table_metadata/extra/", methods=["GET"])
     @protect()
     @statsd_metrics
     @event_logger.log_this_with_context(
@@ -973,7 +1053,8 @@ def select_star(
         self.incr_stats("init", self.select_star.__name__)
         try:
             result = database.select_star(
-                table_name, schema_name, latest_partition=True
+                Table(table_name, schema_name),
+                latest_partition=True,
             )
         except NoSuchTableError:
             self.incr_stats("error", self.select_star.__name__)