rusackas commented on code in PR #38174: URL: https://github.com/apache/superset/pull/38174#discussion_r3365938505
########## superset/db_engine_specs/odps.py: ########## @@ -0,0 +1,192 @@ +# 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 __future__ import annotations + +import logging +from typing import Any, TYPE_CHECKING + +from sqlalchemy import select, text +from sqlalchemy.engine import Dialect + +from superset.databases.schemas import ( + TableMetadataColumnsResponse, + TableMetadataResponse, +) +from superset.databases.utils import ( + get_col_type, + get_foreign_keys_metadata, + get_indexes_metadata, +) +from superset.db_engine_specs.base import BaseEngineSpec, BasicParametersMixin +from superset.sql.parse import Partition, SQLScript, Table +from superset.superset_typing import ResultSetColumnType + +if TYPE_CHECKING: + from superset.models.core import Database + +logger = logging.getLogger(__name__) + + +class OdpsBaseEngineSpec(BaseEngineSpec): + @classmethod + def get_table_metadata( + cls, + database: Database, + table: Table, + partition: Partition | None = None, + ) -> TableMetadataResponse: + """ + Returns basic table metadata + :param database: Database instance + :param table: A Table instance + :param partition: A Table partition info + :return: Basic table metadata + """ + raise NotImplementedError + + +class OdpsEngineSpec(BasicParametersMixin, OdpsBaseEngineSpec): + engine = "odps" + engine_name = "ODPS (MaxCompute)" + default_driver = "odps" + + @classmethod + def get_table_metadata( + cls, database: Any, table: Table, partition: Partition | None = None + ) -> TableMetadataResponse: + """ + Get table metadata information, including type, pk, fks. + This function raises SQLAlchemyError when a schema is not found. + + :param partition: The table's partition info + :param database: The database model + :param table: Table instance + :return: Dict table metadata ready for API response + """ + keys = [] + columns = database.get_columns(table) + primary_key = database.get_pk_constraint(table) + if primary_key and primary_key.get("constrained_columns"): + primary_key["column_names"] = primary_key.pop("constrained_columns") + primary_key["type"] = "pk" + keys += [primary_key] + foreign_keys = get_foreign_keys_metadata(database, table) + indexes = get_indexes_metadata(database, table) + keys += foreign_keys + indexes + payload_columns: list[TableMetadataColumnsResponse] = [] + table_comment = database.get_table_comment(table) + for col in columns: + dtype = get_col_type(col) + payload_columns.append( + { + "name": col["column_name"], + "type": dtype.split("(")[0] if "(" in dtype else dtype, + "longType": dtype, + "keys": [ + k for k in keys if col["column_name"] in k["column_names"] + ], + "comment": col.get("comment"), + } + ) + + with database.get_sqla_engine( + catalog=table.catalog, schema=table.schema + ) as engine: + return { + "name": table.table, + "columns": payload_columns, + "selectStar": cls.select_star( + database=database, + table=table, + dialect=engine.dialect, + limit=100, + show_cols=False, + indent=True, + latest_partition=True, + cols=columns, + partition=partition, + ), + "primaryKey": primary_key, + "foreignKeys": foreign_keys, + "indexes": keys, + "comment": table_comment, + } + + @classmethod + def select_star( # pylint: disable=too-many-arguments + cls, + database: Database, + table: Table, + dialect: Dialect, + limit: int = 100, + show_cols: bool = False, + indent: bool = True, + latest_partition: bool = True, + cols: list[ResultSetColumnType] | None = None, + partition: Partition | None = None, + ) -> str: + """ + Generate a "SELECT * from [schema.]table_name" query with appropriate limit. + + WARNING: expects only unquoted table and schema names. + + :param partition: The table's partition info + :param database: Database instance + :param table: Table instance + :param dialect: SqlAlchemy Dialect instance + :param limit: limit to impose on query + :param show_cols: Show columns in query; otherwise use "*" + :param indent: Add indentation to query + :param latest_partition: Only query the latest partition + :param cols: Columns to include in query + :return: SQL query + """ + # pylint: disable=redefined-outer-name + fields: str | list[Any] = "*" + cols = cols or [] + if (show_cols or latest_partition) and not cols: + cols = database.get_columns(table) + + if show_cols: + fields = cls._get_fields(cols) + full_table_name = cls.quote_table(table, dialect) + qry = select(fields).select_from(text(full_table_name)) + if database.backend == "odps": + if ( + partition is not None + and partition.is_partitioned_table + and partition.partition_column is not None + and len(partition.partition_column) > 0 + ): + partition_str = partition.partition_column[0] + partition_str_where = f"CAST({partition_str} AS STRING) LIKE '%'" + qry = qry.where(text(partition_str_where)) Review Comment: Good catch. Quoted the partition column via `dialect.identifier_preparer.quote(...)` before interpolating it into the `text()` predicate, and added an inline comment explaining why the match-all `LIKE '%'` predicate is used (it forces ODPS to scan a partitioned table, which it otherwise rejects for an unpartitioned full-table preview). Pushed in e35ef6a. ########## superset/daos/database.py: ########## @@ -249,6 +261,69 @@ def get_datasets( .all() ) + @classmethod + def is_odps_partitioned_table( + cls, database: Database, table_name: str + ) -> tuple[bool, list[str]]: + """ + This function is used to determine and retrieve + partition information of the ODPS table. + The return values are whether the partition + table is partitioned and the names of all partition fields. + """ + if not database: + raise ValueError("Database not found") + if database.backend != "odps": + return False, [] + if ODPS is None: + logger.warning("pyodps is not installed, cannot check ODPS partition info") + return False, [] + uri = database.sqlalchemy_uri + access_key = database.password + pattern = re.compile( + r"odps://(?P<username>[^:]+):(?P<password>[^@]+)@(?P<project>[^/]+)/(?:\?" + r"endpoint=(?P<endpoint>[^&]+))" + ) + if not uri or not isinstance(uri, str): + logger.warning( + "Invalid or missing sqlalchemy URI, please provide a correct URI" + ) + return False, [] + if match := pattern.match(unquote(uri)): + access_id = match.group("username") + project = match.group("project") + endpoint = match.group("endpoint") + # `get_table` is a synchronous network call. Bound it with a + # configurable connect/read timeout so an unreachable or slow ODPS + # endpoint can't block the worker indefinitely. + timeout = app.config.get("ODPS_PARTITION_DETECT_TIMEOUT", 30) Review Comment: Done - switched to `app.config["ODPS_PARTITION_DETECT_TIMEOUT"]` so it relies on the config default rather than a duplicated fallback. Pushed in e35ef6a. ########## superset/db_engine_specs/odps.py: ########## @@ -0,0 +1,192 @@ +# 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 __future__ import annotations + +import logging +from typing import Any, TYPE_CHECKING + +from sqlalchemy import select, text +from sqlalchemy.engine import Dialect + +from superset.databases.schemas import ( + TableMetadataColumnsResponse, + TableMetadataResponse, +) +from superset.databases.utils import ( + get_col_type, + get_foreign_keys_metadata, + get_indexes_metadata, +) +from superset.db_engine_specs.base import BaseEngineSpec, BasicParametersMixin +from superset.sql.parse import Partition, SQLScript, Table +from superset.superset_typing import ResultSetColumnType + +if TYPE_CHECKING: + from superset.models.core import Database + +logger = logging.getLogger(__name__) + + +class OdpsBaseEngineSpec(BaseEngineSpec): + @classmethod + def get_table_metadata( + cls, + database: Database, + table: Table, + partition: Partition | None = None, + ) -> TableMetadataResponse: + """ + Returns basic table metadata + :param database: Database instance + :param table: A Table instance + :param partition: A Table partition info + :return: Basic table metadata + """ + raise NotImplementedError + + +class OdpsEngineSpec(BasicParametersMixin, OdpsBaseEngineSpec): + engine = "odps" + engine_name = "ODPS (MaxCompute)" + default_driver = "odps" + + @classmethod + def get_table_metadata( + cls, database: Any, table: Table, partition: Partition | None = None Review Comment: Done - typed `database` as `Database` to match the base spec. Pushed in e35ef6a. (Had to add a couple of narrow casts in the body, since `get_columns`/`get_pk_constraint` return loosely-typed dicts that the shared `databases/utils` helpers also treat as `Any`.) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
