chore: Refactor [Presto|Trino]EngineSpec._partition_query (#23944)

superset/db_engine_specs/hive.py

@@ -468,7 +468,8 @@ class HiveEngineSpec(PrestoEngineSpec):
         order_by: Optional[List[Tuple[str, bool]]] = None,
         filters: Optional[Dict[Any, Any]] = None,
     ) -> str:
-        return f"SHOW PARTITIONS {table_name}"
+        full_table_name = f"{schema}.{table_name}" if schema else table_name
+        return f"SHOW PARTITIONS {full_table_name}"
 
     @classmethod
     def select_star(  # pylint: disable=too-many-arguments

superset/db_engine_specs/presto.py

@@ -464,15 +464,19 @@ class PrestoBaseEngineSpec(BaseEngineSpec, metaclass=ABCMeta):
                 l.append(f"{field} = '{value}'")
             where_clause = "WHERE " + " AND ".join(l)
 
-        presto_version = database.get_extra().get("version")
-
         # Partition select syntax changed in v0.199, so check here.
         # Default to the new syntax if version is unset.
-        partition_select_clause = (
-            f'SELECT * FROM "{table_name}$partitions"'
-            if not presto_version or Version(presto_version) >= Version("0.199")
-            else f"SHOW PARTITIONS FROM {table_name}"
-        )
+        presto_version = database.get_extra().get("version")
+
+        if presto_version and Version(presto_version) < Version("0.199"):
+            full_table_name = f"{schema}.{table_name}" if schema else table_name
+            partition_select_clause = f"SHOW PARTITIONS FROM {full_table_name}"
+        else:
+            system_table_name = f'"{table_name}$partitions"'
+            full_table_name = (
+                f"{schema}.{system_table_name}" if schema else system_table_name
+            )
+            partition_select_clause = f"SELECT * FROM {full_table_name}"
 
         sql = dedent(
             f"""\
@@ -1222,11 +1226,7 @@ class PrestoEngineSpec(PrestoBaseEngineSpec):
                 "cols": sorted(indexes[0].get("column_names", [])),
                 "latest": dict(zip(col_names, latest_parts)),
                 "partitionQuery": cls._partition_query(
-                    table_name=(
-                        f"{schema_name}.{table_name}"
-                        if schema_name and "." not in table_name
-                        else table_name
-                    ),
+                    table_name=table_name,
                     schema=schema_name,
                     indexes=indexes,
                     database=database,

superset/db_engine_specs/trino.py

@@ -79,11 +79,7 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
                 ),
                 "latest": dict(zip(col_names, latest_parts)),
                 "partitionQuery": cls._partition_query(
-                    table_name=(
-                        f"{schema_name}.{table_name}"
-                        if schema_name and "." not in table_name
-                        else table_name
-                    ),
+                    table_name=table_name,
                     schema=schema_name,
                     indexes=indexes,
                     database=database,