diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py
index c1ab44e511..93b1f44186 100644
--- a/superset/connectors/sqla/models.py
+++ b/superset/connectors/sqla/models.py
@@ -530,7 +530,11 @@ class SqlaTable(Model, BaseDatasource):
         # 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
+            self.table_name,
+            sql=self.sql,
+            schema=self.schema,
+            show_cols=False,
+            latest_partition=False,
         )
 
     def get_col(self, col_name: str) -> Optional[Column]:
diff --git a/superset/db_engine_specs/base.py b/superset/db_engine_specs/base.py
index 086a2c907b..0f3375831a 100644
--- a/superset/db_engine_specs/base.py
+++ b/superset/db_engine_specs/base.py
@@ -617,6 +617,7 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         database,
         table_name: str,
         engine: Engine,
+        sql: Optional[str] = None,
         schema: Optional[str] = None,
         limit: int = 100,
         show_cols: bool = False,
@@ -629,6 +630,7 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
 
         :param database: Database instance
         :param table_name: Table name
+        :param sql: SQL defining a subselect
         :param engine: SqlALchemy Engine instance
         :param schema: Schema
         :param limit: limit to impose on query
@@ -638,20 +640,23 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         :param cols: Columns to include in query
         :return: SQL query
         """
-        fields = "*"
-        cols = cols or []
-        if (show_cols or latest_partition) and not cols:
-            cols = database.get_columns(table_name, schema)
-
-        if show_cols:
-            fields = cls._get_fields(cols)
         quote = engine.dialect.identifier_preparer.quote
         if schema:
             full_table_name = quote(schema) + "." + quote(table_name)
         else:
             full_table_name = quote(table_name)
 
-        qry = select(fields).select_from(text(full_table_name))
+        if sql is not None:
+            subselect = f"(\n{sql}\n) AS {quote(table_name)}"
+            qry = select("*").select_from(text(subselect))
+        else:
+            fields = "*"
+            cols = cols or []
+            if (show_cols or latest_partition) and not cols:
+                cols = database.get_columns(table_name, schema)
+            if show_cols:
+                fields = cls._get_fields(cols)
+            qry = select(fields).select_from(text(full_table_name))
 
         if limit:
             qry = qry.limit(limit)
@@ -661,10 +666,10 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
             )
             if partition_query is not None:
                 qry = partition_query
-        sql = database.compile_sqla_query(qry)
+        select_star_query = database.compile_sqla_query(qry)
         if indent:
-            sql = sqlparse.format(sql, reindent=True)
-        return sql
+            select_star_query = sqlparse.format(select_star_query, reindent=True)
+        return select_star_query
 
     @classmethod
     def estimate_statement_cost(
diff --git a/superset/db_engine_specs/hive.py b/superset/db_engine_specs/hive.py
index 7c680cb00c..fd2c85dbde 100644
--- a/superset/db_engine_specs/hive.py
+++ b/superset/db_engine_specs/hive.py
@@ -354,6 +354,7 @@ class HiveEngineSpec(PrestoEngineSpec):
         database,
         table_name: str,
         engine: Engine,
+        sql: Optional[str] = None,
         schema: str = None,
         limit: int = 100,
         show_cols: bool = False,
@@ -367,6 +368,7 @@ class HiveEngineSpec(PrestoEngineSpec):
             database,
             table_name,
             engine,
+            sql,
             schema,
             limit,
             show_cols,
diff --git a/superset/db_engine_specs/presto.py b/superset/db_engine_specs/presto.py
index 9f91969d0e..661fe3517a 100644
--- a/superset/db_engine_specs/presto.py
+++ b/superset/db_engine_specs/presto.py
@@ -409,6 +409,7 @@ class PrestoEngineSpec(BaseEngineSpec):
         database,
         table_name: str,
         engine: Engine,
+        sql: Optional[str] = None,
         schema: str = None,
         limit: int = 100,
         show_cols: bool = False,
@@ -432,6 +433,7 @@ class PrestoEngineSpec(BaseEngineSpec):
             database,
             table_name,
             engine,
+            sql,
             schema,
             limit,
             show_cols,
diff --git a/superset/models/core.py b/superset/models/core.py
index ccc909af08..d378248edc 100755
--- a/superset/models/core.py
+++ b/superset/models/core.py
@@ -1006,6 +1006,7 @@ class Database(Model, AuditMixinNullable, ImportMixin):
     def select_star(
         self,
         table_name: str,
+        sql: Optional[str] = None,
         schema: Optional[str] = None,
         limit: int = 100,
         show_cols: bool = False,
@@ -1020,6 +1021,7 @@ class Database(Model, AuditMixinNullable, ImportMixin):
         return self.db_engine_spec.select_star(
             self,
             table_name,
+            sql=sql,
             schema=schema,
             engine=eng,
             limit=limit,