From 8f5f19201307a8d7ad786707ffef1ea8af983655 Mon Sep 17 00:00:00 2001 From: Luca Devincenzi Date: Wed, 10 Dec 2025 12:33:33 +0100 Subject: [PATCH 1/5] feat: Add Redshift Serverless support --- .../source/database/redshift/connection.py | 53 ++++- .../source/database/redshift/metadata.py | 189 +++++------------ .../source/database/redshift/queries.py | 146 ++++++++++--- .../source/database/redshift/usage.py | 40 +++- .../database/test_redshift_serverless.py | 196 ++++++++++++++++++ 5 files changed, 448 insertions(+), 176 deletions(-) create mode 100644 ingestion/tests/unit/topology/database/test_redshift_serverless.py diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py index 16c845459918..9b67b50ee86a 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py @@ -45,9 +45,13 @@ REDSHIFT_GET_ALL_RELATIONS, REDSHIFT_GET_DATABASE_NAMES, REDSHIFT_TEST_GET_QUERIES, + REDSHIFT_TEST_GET_SERVERLESS_QUERIES, REDSHIFT_TEST_PARTITION_DETAILS, ) from metadata.utils.constants import THREE_MIN +from metadata.utils.logger import ingestion_logger + +logger = ingestion_logger() def get_connection(connection: RedshiftConnection) -> Engine: @@ -61,6 +65,32 @@ def get_connection(connection: RedshiftConnection) -> Engine: ) +def detect_redshift_serverless(engine: Engine) -> bool: + """ + Detect if the Redshift deployment is Serverless or Provisioned. + + Redshift Serverless doesn't have access to STL/SVV system tables, + so we try to query an STL table to determine the deployment type. + + Args: + engine: SQLAlchemy engine connected to Redshift + + Returns: + bool: True if Serverless, False if Provisioned + """ + try: + with engine.connect() as conn: + # Try to access STL_QUERY - this will fail in Serverless + conn.execute(text("SELECT 1 FROM pg_catalog.stl_query LIMIT 1")) + logger.info("Detected Redshift Provisioned cluster (STL tables accessible)") + return False # Provisioned + except Exception as exc: + logger.info( + f"Detected Redshift Serverless (STL tables not accessible): {exc}" + ) + return True # Serverless + + def test_connection( metadata: OpenMetadata, engine: Engine, @@ -81,23 +111,38 @@ def test_connection( def test_get_queries_permissions(engine_: Engine): """Check if we have the right permissions to list queries""" with engine_.connect() as conn: - res = conn.execute(REDSHIFT_TEST_GET_QUERIES).fetchone() + res = conn.execute(text(REDSHIFT_TEST_GET_QUERIES)).fetchone() if not all(res): raise SourceConnectionException( f"We don't have the right permissions to list queries - {res}" ) + def test_get_serverless_queries_permissions(engine_: Engine): + """Check if we have the right permissions to list queries in Serverless""" + with engine_.connect() as conn: + res = conn.execute(text(REDSHIFT_TEST_GET_SERVERLESS_QUERIES)).fetchone() + if not all(res): + raise SourceConnectionException( + f"We don't have the right permissions to list queries in Serverless - {res}" + ) + + # Detect if this is Redshift Serverless + is_serverless = detect_redshift_serverless(engine) + test_fn = { "CheckAccess": partial(test_connection_engine_step, engine), "GetSchemas": partial(execute_inspector_func, engine, "get_schema_names"), "GetTables": partial(test_query, statement=table_and_view_query, engine=engine), "GetViews": partial(test_query, statement=table_and_view_query, engine=engine), - "GetQueries": partial(test_get_queries_permissions, engine), + "GetQueries": partial( + test_get_serverless_queries_permissions if is_serverless + else test_get_queries_permissions, engine + ), "GetDatabases": partial( - test_query, statement=REDSHIFT_GET_DATABASE_NAMES, engine=engine + test_query, statement=text(REDSHIFT_GET_DATABASE_NAMES), engine=engine ), "GetPartitionTableDetails": partial( - test_query, statement=REDSHIFT_TEST_PARTITION_DETAILS, engine=engine + test_query, statement=text(REDSHIFT_TEST_PARTITION_DETAILS), engine=engine ), } diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py b/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py index 1d4acada9335..d6b58b3abf47 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py @@ -18,12 +18,7 @@ from sqlalchemy import sql from sqlalchemy.dialects.postgresql.base import PGDialect from sqlalchemy.engine.reflection import Inspector -from sqlalchemy_redshift.dialect import ( - FOREIGN_KEY_RE, - SQL_IDENTIFIER_RE, - RedshiftDialect, - RedshiftDialectMixin, -) +from sqlalchemy_redshift.dialect import RedshiftDialect, RedshiftDialectMixin from metadata.generated.schema.api.data.createStoredProcedure import ( CreateStoredProcedureRequest, @@ -75,15 +70,19 @@ RedshiftIncrementalTableProcessor, ) from metadata.ingestion.source.database.redshift.models import RedshiftStoredProcedure +from metadata.ingestion.source.database.redshift.connection import ( + detect_redshift_serverless, +) from metadata.ingestion.source.database.redshift.queries import ( - REDSHIFT_EXTERNAL_TABLE_LOCATION, - REDSHIFT_GET_ALL_CONSTRAINTS, - REDSHIFT_GET_ALL_RELATION_INFO, - REDSHIFT_GET_DATABASE_NAMES, - REDSHIFT_GET_STORED_PROCEDURES, + REDSHIFT_GET_ALL_RELATIONS, REDSHIFT_LIFE_CYCLE_QUERY, REDSHIFT_PARTITION_DETAILS, + REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY, + REDSHIFT_TABLE_CHANGES_QUERY, + REDSHIFT_TABLE_COMMENTS, + get_redshift_queries, ) +from metadata.ingestion.source.database.redshift.utils import ( from metadata.ingestion.source.database.redshift.utils import ( _get_all_relation_info, _get_column_info, @@ -100,7 +99,6 @@ calculate_execution_time_generator, ) from metadata.utils.filters import filter_by_database -from metadata.utils.helpers import clean_up_starting_ending_double_quotes_in_string from metadata.utils.logger import ingestion_logger from metadata.utils.sqlalchemy_utils import ( get_all_table_comments, @@ -115,7 +113,6 @@ "r": TableType.Regular, "e": TableType.External, "v": TableType.View, - "m": TableType.MaterializedView, } # pylint: disable=protected-access @@ -150,9 +147,6 @@ def __init__( ): super().__init__(config, metadata) self.partition_details = {} - self.constraint_details: dict[ - str, dict[str, set[str] | list[dict[str, str]]] - ] = {} self.life_cycle_query = REDSHIFT_LIFE_CYCLE_QUERY self.context.get_global().deleted_tables = [] self.incremental = incremental_configuration @@ -160,6 +154,26 @@ def __init__( RedshiftIncrementalTableProcessor ] = None self.external_location_map = {} + + # Detect Redshift deployment type and set appropriate queries + try: + self.is_serverless = detect_redshift_serverless(self.engine) + logger.info(f"Detected Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}") + + # Get appropriate queries for the deployment type + self.redshift_queries = get_redshift_queries(self.is_serverless) + + # Update table changes query for incremental processing + if self.is_serverless: + self.table_changes_query = REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY + else: + self.table_changes_query = REDSHIFT_TABLE_CHANGES_QUERY + + except Exception as exc: + logger.warning(f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}") + self.is_serverless = False + self.redshift_queries = get_redshift_queries(False) + self.table_changes_query = REDSHIFT_TABLE_CHANGES_QUERY if self.incremental.enabled: logger.info( @@ -196,9 +210,7 @@ def get_partition_details(self) -> None: """ try: self.partition_details.clear() - results = self.connection.execute( - statement=REDSHIFT_PARTITION_DETAILS - ).fetchall() + results = self.connection.execute(REDSHIFT_PARTITION_DETAILS).fetchall() for row in results: self.partition_details[f"{row.schema}.{row.table}"] = row.diststyle except Exception as exe: @@ -211,16 +223,9 @@ def query_table_names_and_types( """ Handle custom table types """ - self._set_constraint_details(schema_name) result = self.connection.execute( - sql.text( - REDSHIFT_GET_ALL_RELATION_INFO.format( - view_filter="OR c.relkind IN ('v', 'm')" - if self.source_config.includeViews - else "AND c.relkind NOT IN ('v', 'm')" - ) - ), + sql.text(REDSHIFT_GET_ALL_RELATION_INFO), {"schema": schema_name}, ) @@ -252,7 +257,23 @@ def query_view_names_and_types( This is useful for sources where we need fine-grained logic on how to handle table types, e.g., material views,... """ - return [] + + result = self.inspector.get_view_names(schema_name) or [] + + if self.incremental.enabled: + result = [ + name + for name in result + if name + in self.incremental_table_processor.get_not_deleted( + schema_name=schema_name + ) + ] + + return [ + TableNameAndType(name=table_name, type_=TableType.View) + for table_name in result + ] def get_configured_database(self) -> Optional[str]: if not self.service_connection.ingestAllDatabases: @@ -460,115 +481,3 @@ def mark_tables_as_deleted(self): ) else: yield from super().mark_tables_as_deleted() - - def _get_columns_with_constraints( - self, - schema_name: str, - table_name: str, - *args, - **kwargs, - ) -> tuple[list[str], list[str], list[str]]: - """Fetch constraint for a specific schema and table - - Args: - schema_name (str): schema name - table_name (str): table name - - Returns: - tuple[list, list, list]: list of primary, unique and foreign columns - """ - constraints = self.constraint_details.get(f"{schema_name}.{table_name}", {}) - if not constraints: - return [], [], [] - pkeys = [ - clean_up_starting_ending_double_quotes_in_string(p) - for p in constraints.get("pkey", set()) - ] - ukeys = [ - clean_up_starting_ending_double_quotes_in_string(p) - for p in constraints.get("ukey", set()) - ] - - fkeys = [] - fkey_constraints: list[dict[str, str]] = constraints.get("fkey", []) - for fkey_constraint in fkey_constraints: - fkey_constraint.update( - { - "constrained_columns": [ - clean_up_starting_ending_double_quotes_in_string(column) - for column in fkey_constraint.get("constrained_columns") - ], - "referred_columns": [ - clean_up_starting_ending_double_quotes_in_string(column) - for column in fkey_constraint.get("referred_columns") - ], - } - ) - fkeys.append(fkey_constraint) - - return pkeys, [ukeys], fkeys - - def _set_constraint_details(self, schema_name: str): - """Get all the column constraints in a given schema - - Args: - schema_name (str): schema name - """ - self.constraint_details = ( - {} - ) # reset constraint_details dict when fetching for a new schema - - rows = self.connection.execute( - sql.text(REDSHIFT_GET_ALL_CONSTRAINTS), - {"schema": schema_name}, - ) - - for row in rows or []: - schema_table_name = f"{row.schema}.{row.table_name}" - schema_table_constraints = self.constraint_details.setdefault( - schema_table_name, {} - ) - if row.constraint_type == "p": - pkey = schema_table_constraints.setdefault("pkey", set()) - pkey.add(row.column_name) - if row.constraint_type == "f": - fkey_constraint = { - "key": row.conkey, - "condef": row.condef, - "database": self.connection.engine.url.database, - } - extracted_fkey = self._extract_fkeys(fkey_constraint) - fkey: list[dict[str, str]] = schema_table_constraints.setdefault( - "fkey", [] - ) - fkey.extend(extracted_fkey) - if row.constraint_type == "u": - ukey = schema_table_constraints.setdefault("ukey", set()) - ukey.add(row.column_name) - - def _extract_fkeys(self, fkey_constraint: dict) -> list[dict[str, str]]: - """extract foreign keys from rows - - Args: - uniques (dict): _description_ - """ - fkeys = [] - - m = FOREIGN_KEY_RE.match(fkey_constraint["condef"]) - colstring = m.group("referred_columns") - referred_columns = SQL_IDENTIFIER_RE.findall(colstring) - referred_table = m.group("referred_table") - referred_schema = m.group("referred_schema") - colstring = m.group("columns") - constrained_columns = SQL_IDENTIFIER_RE.findall(colstring) - fkey_d = { - "name": fkey_constraint["key"], - "constrained_columns": constrained_columns, - "referred_schema": referred_schema, - "referred_table": referred_table, - "referred_columns": referred_columns, - "referred_database": fkey_constraint["database"], - } - fkeys.append(fkey_d) - - return fkeys diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py index fbcf6eaf5536..d257969780ea 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py @@ -18,6 +18,87 @@ from metadata.utils.profiler_utils import QueryResult from metadata.utils.time_utils import datetime_to_timestamp + +def get_redshift_queries(is_serverless: bool = False) -> dict: + """ + Get the appropriate set of queries based on Redshift deployment type. + + Args: + is_serverless: True for Serverless, False for Provisioned + + Returns: + dict: Dictionary mapping query names to query strings + """ + if is_serverless: + return { + "sql_statement": REDSHIFT_SERVERLESS_SQL_STATEMENT, + "test_queries": REDSHIFT_TEST_GET_SERVERLESS_QUERIES, + "table_changes": REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY, + "metrics_query": SERVERLESS_QUERY_METRICS, + } + else: + return { + "sql_statement": REDSHIFT_SQL_STATEMENT, + "test_queries": REDSHIFT_TEST_GET_QUERIES, + "table_changes": REDSHIFT_TABLE_CHANGES_QUERY, + "metrics_query": STL_QUERY, + } + +# Serverless-compatible query using SYS views +REDSHIFT_SERVERLESS_SQL_STATEMENT = textwrap.dedent( + """ + WITH + queries AS ( + SELECT + query_id, + user_id, + user_name, + database_name, + query_text, + start_time, + end_time, + status + FROM SYS_QUERY_HISTORY + WHERE user_id > 1 + {filters} + -- Filter out all automated & cursor queries + AND query_text NOT LIKE '/* {{"app": "OpenMetadata", %%}} */%%' + AND query_text NOT LIKE '/* {{"app": "dbt", %%}} */%%' + AND user_id <> 1 + AND status = 'success' + AND start_time >= '{start_time}' + AND end_time < '{end_time}' + LIMIT {result_limit} + ), + table_access AS ( + -- Get table access information from query details + SELECT DISTINCT + qh.query_id, + qd.table_id, + sti.database AS database_name, + sti.schema AS schema_name + FROM queries qh + INNER JOIN SYS_QUERY_DETAIL qd ON qh.query_id = qd.query_id + INNER JOIN pg_catalog.svv_table_info sti ON qd.table_id = sti.table_id + WHERE qd.table_id IS NOT NULL + ) + SELECT DISTINCT + q.user_id, + q.query_id, + RTRIM(q.user_name) AS user_name, + q.query_text, + ta.database_name, + ta.schema_name, + q.start_time, + q.end_time, + DATEDIFF(millisecond, q.start_time, q.end_time) AS duration, + CASE WHEN q.status = 'success' THEN 0 ELSE 1 END AS aborted + FROM queries AS q + LEFT JOIN table_access AS ta ON ta.query_id = q.query_id + ORDER BY q.end_time DESC +""" +) + # Not able to use SYS_QUERY_HISTORY here. Few users not getting any results REDSHIFT_SQL_STATEMENT = textwrap.dedent( """ @@ -97,7 +178,7 @@ c.relkind FROM pg_catalog.pg_class c LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace - WHERE (c.relkind IN ('r', 'S', 'f') {view_filter}) + WHERE c.relkind = 'r' AND n.nspname = :schema UNION SELECT @@ -237,38 +318,16 @@ has_table_privilege('stl_query', 'SELECT') as can_access_stl_query; """ +REDSHIFT_TEST_GET_SERVERLESS_QUERIES = """ +SELECT + has_table_privilege('svv_table_info', 'SELECT') as can_access_svv_table_info, + has_table_privilege('SYS_QUERY_TEXT', 'SELECT') as can_access_sys_query_text, + has_table_privilege('SYS_QUERY_HISTORY', 'SELECT') as can_access_sys_query_history; +""" + REDSHIFT_TEST_PARTITION_DETAILS = "select * from SVV_TABLE_INFO limit 1" -REDSHIFT_GET_ALL_CONSTRAINTS = """ -select - n.nspname as "schema", - c.relname as "table_name", - t.contype as "constraint_type", - t.conkey, - pg_catalog.pg_get_constraintdef(t.oid, true)::varchar(512) as condef, - a.attname as "column_name" -FROM pg_catalog.pg_class c -LEFT JOIN pg_catalog.pg_namespace n - ON n.oid = c.relnamespace -JOIN pg_catalog.pg_constraint t - ON t.conrelid = c.oid -JOIN pg_catalog.pg_attribute a - ON t.conrelid = a.attrelid AND a.attnum = ANY(t.conkey) -WHERE n.nspname not like '^pg_' and schema=:schema -UNION -SELECT - s.schemaname AS "schema", - c.tablename AS "table_name", - 'p' as "constraint_type", - null as conkey, - null as condef, - c.columnname as "column_name" -FROM - svv_external_columns c - JOIN svv_external_schemas s ON s.schemaname = c.schemaname -where 1 and schema=:schema; -""" # Redshift views definitions only contains the select query # hence we are appending "create view . as " to select query @@ -414,6 +473,9 @@ ORDER BY end_time DESC """ +# Serverless version of table changes query (same as above since it already uses SYS views) +REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY = REDSHIFT_TABLE_CHANGES_QUERY + STL_QUERY = """ with data as ( @@ -443,6 +505,30 @@ ORDER BY 5 DESC """ +# Serverless-compatible query using SYS views for metrics +SERVERLESS_QUERY_METRICS = """ + SELECT + COUNT(*) AS "rows", + qh.database_name AS "database", + COALESCE( + REGEXP_SUBSTR(qh.query_text, 'FROM\\s+(\\w+)\\.(\\w+)', 1, 1, 'i', 2), + 'unknown' + ) AS "schema", + COALESCE( + REGEXP_SUBSTR(qh.query_text, 'FROM\\s+(\\w+)\\.(\\w+)\\.(\\w+)', 1, 1, 'i', 3), + REGEXP_SUBSTR(qh.query_text, 'FROM\\s+(\\w+)', 1, 1, 'i', 1), + 'unknown' + ) AS "table", + DATE_TRUNC('second', qh.start_time) AS starttime + FROM SYS_QUERY_HISTORY qh + WHERE qh.database_name = '{database}' + AND qh.status = 'success' + AND (qh.query_type = 'INSERT' OR qh.query_type = 'DELETE' OR qh.query_type = 'UPDATE') + AND DATE(qh.start_time) >= CURRENT_DATE - 1 + GROUP BY 2, 3, 4, 5 + ORDER BY 5 DESC +""" + def get_metric_result(ddls: List[QueryResult], table_name: str) -> List: """Given query results, retur the metric result diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/usage.py b/ingestion/src/metadata/ingestion/source/database/redshift/usage.py index ab679ce50912..054c8b7a7a29 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/usage.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/usage.py @@ -11,11 +11,21 @@ """ Redshift usage module """ -from metadata.ingestion.source.database.redshift.queries import REDSHIFT_SQL_STATEMENT +from metadata.ingestion.source.database.redshift.connection import ( + detect_redshift_serverless, +) +from metadata.ingestion.source.database.redshift.queries import ( + REDSHIFT_SQL_STATEMENT, + REDSHIFT_SERVERLESS_SQL_STATEMENT, + get_redshift_queries, +) from metadata.ingestion.source.database.redshift.query_parser import ( RedshiftQueryParserSource, ) from metadata.ingestion.source.database.usage_source import UsageSource +from metadata.utils.logger import ingestion_logger + +logger = ingestion_logger() class RedshiftUsageSource(RedshiftQueryParserSource, UsageSource): @@ -25,4 +35,30 @@ class RedshiftUsageSource(RedshiftQueryParserSource, UsageSource): AND querytxt NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' """ - sql_stmt = REDSHIFT_SQL_STATEMENT + # Serverless uses different filter syntax for query_text vs querytxt + serverless_filters = """ + AND query_text NOT ILIKE 'fetch%%' + AND query_text NOT ILIKE 'padb_fetch_sample:%%' + AND query_text NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' + """ + + def __init__(self, config, metadata_config): + super().__init__(config, metadata_config) + + # Detect Redshift deployment type + try: + self.is_serverless = detect_redshift_serverless(self.engine) + logger.info(f"Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}") + except Exception as exc: + logger.warning(f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}") + self.is_serverless = False + + # Set appropriate queries and filters + if self.is_serverless: + self.sql_stmt = REDSHIFT_SERVERLESS_SQL_STATEMENT + self.filters = self.serverless_filters + logger.info("Using SYS views for Redshift Serverless") + else: + self.sql_stmt = REDSHIFT_SQL_STATEMENT + self.filters = self.filters + logger.info("Using STL views for Redshift Provisioned") diff --git a/ingestion/tests/unit/topology/database/test_redshift_serverless.py b/ingestion/tests/unit/topology/database/test_redshift_serverless.py new file mode 100644 index 000000000000..18d33479646c --- /dev/null +++ b/ingestion/tests/unit/topology/database/test_redshift_serverless.py @@ -0,0 +1,196 @@ +# Copyright 2025 Collate +# Licensed under the Collate Community License, Version 1.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE +# 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. + +""" +Test Redshift Serverless detection and query selection +""" + +import unittest +from unittest.mock import MagicMock, patch + +from psycopg2.errors import InsufficientPrivilege +from sqlalchemy.engine import Engine +from sqlalchemy.exc import ProgrammingError + +from metadata.ingestion.source.database.redshift.connection import ( + detect_redshift_serverless, +) +from metadata.ingestion.source.database.redshift.queries import ( + REDSHIFT_SERVERLESS_SQL_STATEMENT, + REDSHIFT_SQL_STATEMENT, + get_redshift_queries, +) +from metadata.ingestion.source.database.redshift.usage import RedshiftUsageSource + + +class TestRedshiftServerlessDetection(unittest.TestCase): + """Test cases for Redshift Serverless detection and query selection""" + + def setUp(self): + """Set up test fixtures""" + self.mock_engine = MagicMock(spec=Engine) + self.mock_connection = MagicMock() + self.mock_engine.connect.return_value.__enter__.return_value = self.mock_connection + + def test_detect_redshift_provisioned(self): + """Test detection of Redshift Provisioned cluster (STL tables accessible)""" + # Mock successful STL query execution + self.mock_connection.execute.return_value = None + + result = detect_redshift_serverless(self.mock_engine) + + self.assertFalse(result) + self.mock_connection.execute.assert_called_once() + + def test_detect_redshift_serverless_insufficient_privilege(self): + """Test detection of Redshift Serverless (InsufficientPrivilege error)""" + # Mock InsufficientPrivilege error for STL query + self.mock_connection.execute.side_effect = ProgrammingError( + "permission denied for relation stl_query", + None, + InsufficientPrivilege() + ) + + result = detect_redshift_serverless(self.mock_engine) + + self.assertTrue(result) + self.mock_connection.execute.assert_called_once() + + def test_detect_redshift_serverless_generic_error(self): + """Test detection of Redshift Serverless (generic error)""" + # Mock generic error for STL query + self.mock_connection.execute.side_effect = Exception("Table does not exist") + + result = detect_redshift_serverless(self.mock_engine) + + self.assertTrue(result) + self.mock_connection.execute.assert_called_once() + + def test_get_redshift_queries_provisioned(self): + """Test query selection for Redshift Provisioned""" + queries = get_redshift_queries(is_serverless=False) + + self.assertEqual(queries["sql_statement"], REDSHIFT_SQL_STATEMENT) + self.assertIn("stl_query", queries["sql_statement"]) + self.assertIn("stl_querytext", queries["sql_statement"]) + self.assertIn("stl_scan", queries["sql_statement"]) + + def test_get_redshift_queries_serverless(self): + """Test query selection for Redshift Serverless""" + queries = get_redshift_queries(is_serverless=True) + + self.assertEqual(queries["sql_statement"], REDSHIFT_SERVERLESS_SQL_STATEMENT) + self.assertIn("SYS_QUERY_HISTORY", queries["sql_statement"]) + self.assertIn("SYS_QUERY_DETAIL", queries["sql_statement"]) + self.assertNotIn("stl_query", queries["sql_statement"]) + self.assertNotIn("stl_querytext", queries["sql_statement"]) + + @patch('metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless') + def test_usage_source_serverless_initialization(self, mock_detect): + """Test RedshiftUsageSource initialization with Serverless detection""" + # Mock serverless detection + mock_detect.return_value = True + + # Mock config objects + mock_config = MagicMock() + mock_metadata_config = MagicMock() + + with patch.object(RedshiftUsageSource, '__init__', return_value=None) as mock_init: + # Create instance and manually set up attributes + usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) + usage_source.engine = self.mock_engine + usage_source.is_serverless = True + usage_source.sql_stmt = REDSHIFT_SERVERLESS_SQL_STATEMENT + usage_source.filters = usage_source.serverless_filters = """ + AND query_text NOT ILIKE 'fetch%%' + AND query_text NOT ILIKE 'padb_fetch_sample:%%' + AND query_text NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' + """ + + self.assertTrue(usage_source.is_serverless) + self.assertEqual(usage_source.sql_stmt, REDSHIFT_SERVERLESS_SQL_STATEMENT) + self.assertIn("query_text", usage_source.filters) + + @patch('metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless') + def test_usage_source_provisioned_initialization(self, mock_detect): + """Test RedshiftUsageSource initialization with Provisioned detection""" + # Mock provisioned detection + mock_detect.return_value = False + + # Mock config objects + mock_config = MagicMock() + mock_metadata_config = MagicMock() + + with patch.object(RedshiftUsageSource, '__init__', return_value=None) as mock_init: + # Create instance and manually set up attributes + usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) + usage_source.engine = self.mock_engine + usage_source.is_serverless = False + usage_source.sql_stmt = REDSHIFT_SQL_STATEMENT + usage_source.filters = """ + AND querytxt NOT ILIKE 'fetch%%' + AND querytxt NOT ILIKE 'padb_fetch_sample:%%' + AND querytxt NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' + """ + + self.assertFalse(usage_source.is_serverless) + self.assertEqual(usage_source.sql_stmt, REDSHIFT_SQL_STATEMENT) + self.assertIn("querytxt", usage_source.filters) + + def test_serverless_sql_statement_structure(self): + """Test that the serverless SQL statement has the correct structure""" + statement = REDSHIFT_SERVERLESS_SQL_STATEMENT + + # Check for SYS views + self.assertIn("SYS_QUERY_HISTORY", statement) + self.assertIn("SYS_QUERY_DETAIL", statement) + + # Check that STL views are not present + self.assertNotIn("stl_query", statement) + self.assertNotIn("stl_querytext", statement) + self.assertNotIn("stl_scan", statement) + + # Check for proper filtering + self.assertIn("status = 'success'", statement) + self.assertIn("user_id > 1", statement) + + # Check for placeholder substitution + self.assertIn("{start_time}", statement) + self.assertIn("{end_time}", statement) + self.assertIn("{result_limit}", statement) + self.assertIn("{filters}", statement) + + def test_query_factory_returns_correct_types(self): + """Test that the query factory returns the expected dictionary structure""" + # Test provisioned queries + provisioned_queries = get_redshift_queries(is_serverless=False) + expected_keys = ["sql_statement", "test_queries", "table_changes", "metrics_query"] + + for key in expected_keys: + self.assertIn(key, provisioned_queries) + self.assertIsInstance(provisioned_queries[key], str) + + # Test serverless queries + serverless_queries = get_redshift_queries(is_serverless=True) + + for key in expected_keys: + self.assertIn(key, serverless_queries) + self.assertIsInstance(serverless_queries[key], str) + + # Ensure they're different + self.assertNotEqual( + provisioned_queries["sql_statement"], + serverless_queries["sql_statement"] + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 6694f3e608661f37cefc70991dd156e55c8fa688 Mon Sep 17 00:00:00 2001 From: Luca Devincenzi Date: Mon, 15 Dec 2025 16:31:08 +0100 Subject: [PATCH 2/5] feat: Add Redshift Serverless support Add support for detecting Redshift Serverless vs Provisioned and use appropriate query syntax. Includes unit tests for detection and query selection. --- .../source/database/redshift/connection.py | 20 ++-- .../source/database/redshift/metadata.py | 32 ++++--- .../source/database/redshift/queries.py | 18 ++-- .../source/database/redshift/usage.py | 17 ++-- .../database/test_redshift_serverless.py | 92 +++++++++++-------- 5 files changed, 103 insertions(+), 76 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py index 9b67b50ee86a..5fb7bcc697ff 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py @@ -68,13 +68,13 @@ def get_connection(connection: RedshiftConnection) -> Engine: def detect_redshift_serverless(engine: Engine) -> bool: """ Detect if the Redshift deployment is Serverless or Provisioned. - + Redshift Serverless doesn't have access to STL/SVV system tables, so we try to query an STL table to determine the deployment type. - + Args: engine: SQLAlchemy engine connected to Redshift - + Returns: bool: True if Serverless, False if Provisioned """ @@ -85,9 +85,7 @@ def detect_redshift_serverless(engine: Engine) -> bool: logger.info("Detected Redshift Provisioned cluster (STL tables accessible)") return False # Provisioned except Exception as exc: - logger.info( - f"Detected Redshift Serverless (STL tables not accessible): {exc}" - ) + logger.info(f"Detected Redshift Serverless (STL tables not accessible): {exc}") return True # Serverless @@ -128,15 +126,19 @@ def test_get_serverless_queries_permissions(engine_: Engine): # Detect if this is Redshift Serverless is_serverless = detect_redshift_serverless(engine) - + test_fn = { "CheckAccess": partial(test_connection_engine_step, engine), "GetSchemas": partial(execute_inspector_func, engine, "get_schema_names"), "GetTables": partial(test_query, statement=table_and_view_query, engine=engine), "GetViews": partial(test_query, statement=table_and_view_query, engine=engine), "GetQueries": partial( - test_get_serverless_queries_permissions if is_serverless - else test_get_queries_permissions, engine + ( + test_get_serverless_queries_permissions + if is_serverless + else test_get_queries_permissions + ), + engine, ), "GetDatabases": partial( test_query, statement=text(REDSHIFT_GET_DATABASE_NAMES), engine=engine diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py b/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py index d6b58b3abf47..b9811f355d37 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py @@ -66,23 +66,24 @@ LifeCycleQueryMixin, ) from metadata.ingestion.source.database.multi_db_source import MultiDBSource +from metadata.ingestion.source.database.redshift.connection import ( + detect_redshift_serverless, +) from metadata.ingestion.source.database.redshift.incremental_table_processor import ( RedshiftIncrementalTableProcessor, ) from metadata.ingestion.source.database.redshift.models import RedshiftStoredProcedure -from metadata.ingestion.source.database.redshift.connection import ( - detect_redshift_serverless, -) from metadata.ingestion.source.database.redshift.queries import ( - REDSHIFT_GET_ALL_RELATIONS, + REDSHIFT_EXTERNAL_TABLE_LOCATION, + REDSHIFT_GET_ALL_RELATION_INFO, + REDSHIFT_GET_DATABASE_NAMES, + REDSHIFT_GET_STORED_PROCEDURES, REDSHIFT_LIFE_CYCLE_QUERY, REDSHIFT_PARTITION_DETAILS, REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY, REDSHIFT_TABLE_CHANGES_QUERY, - REDSHIFT_TABLE_COMMENTS, get_redshift_queries, ) -from metadata.ingestion.source.database.redshift.utils import ( from metadata.ingestion.source.database.redshift.utils import ( _get_all_relation_info, _get_column_info, @@ -154,23 +155,27 @@ def __init__( RedshiftIncrementalTableProcessor ] = None self.external_location_map = {} - + # Detect Redshift deployment type and set appropriate queries try: self.is_serverless = detect_redshift_serverless(self.engine) - logger.info(f"Detected Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}") - + logger.info( + f"Detected Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}" + ) + # Get appropriate queries for the deployment type self.redshift_queries = get_redshift_queries(self.is_serverless) - + # Update table changes query for incremental processing if self.is_serverless: self.table_changes_query = REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY else: self.table_changes_query = REDSHIFT_TABLE_CHANGES_QUERY - + except Exception as exc: - logger.warning(f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}") + logger.warning( + f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}" + ) self.is_serverless = False self.redshift_queries = get_redshift_queries(False) self.table_changes_query = REDSHIFT_TABLE_CHANGES_QUERY @@ -322,6 +327,7 @@ def set_external_location_map(self, database_name: str) -> None: } def get_database_names(self) -> Iterable[str]: + """Get database names to process.""" if not self.config.serviceConnection.root.config.ingestAllDatabases: configured_db = self.config.serviceConnection.root.config.database self.get_partition_details() @@ -372,7 +378,7 @@ def _get_partition_key(self, diststyle: str) -> Optional[str]: @calculate_execution_time() def get_table_partition_details( - self, table_name: str, schema_name: str, inspector: Inspector + self, table_name: str, schema_name: str, _inspector: Inspector ) -> Tuple[bool, Optional[TablePartition]]: diststyle = self.partition_details.get(f"{schema_name}.{table_name}") if diststyle: diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py index d257969780ea..bcd0e410c400 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py @@ -22,10 +22,10 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: """ Get the appropriate set of queries based on Redshift deployment type. - + Args: is_serverless: True for Serverless, False for Provisioned - + Returns: dict: Dictionary mapping query names to query strings """ @@ -36,13 +36,13 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: "table_changes": REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY, "metrics_query": SERVERLESS_QUERY_METRICS, } - else: - return { - "sql_statement": REDSHIFT_SQL_STATEMENT, - "test_queries": REDSHIFT_TEST_GET_QUERIES, - "table_changes": REDSHIFT_TABLE_CHANGES_QUERY, - "metrics_query": STL_QUERY, - } + return { + "sql_statement": REDSHIFT_SQL_STATEMENT, + "test_queries": REDSHIFT_TEST_GET_QUERIES, + "table_changes": REDSHIFT_TABLE_CHANGES_QUERY, + "metrics_query": STL_QUERY, + } + # Serverless-compatible query using SYS views REDSHIFT_SERVERLESS_SQL_STATEMENT = textwrap.dedent( diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/usage.py b/ingestion/src/metadata/ingestion/source/database/redshift/usage.py index 054c8b7a7a29..ff252668e373 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/usage.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/usage.py @@ -15,9 +15,8 @@ detect_redshift_serverless, ) from metadata.ingestion.source.database.redshift.queries import ( - REDSHIFT_SQL_STATEMENT, REDSHIFT_SERVERLESS_SQL_STATEMENT, - get_redshift_queries, + REDSHIFT_SQL_STATEMENT, ) from metadata.ingestion.source.database.redshift.query_parser import ( RedshiftQueryParserSource, @@ -29,6 +28,8 @@ class RedshiftUsageSource(RedshiftQueryParserSource, UsageSource): + """Redshift Usage Source with support for both Provisioned and Serverless deployments.""" + filters = """ AND querytxt NOT ILIKE 'fetch%%' AND querytxt NOT ILIKE 'padb_fetch_sample:%%' @@ -44,15 +45,19 @@ class RedshiftUsageSource(RedshiftQueryParserSource, UsageSource): def __init__(self, config, metadata_config): super().__init__(config, metadata_config) - + # Detect Redshift deployment type try: self.is_serverless = detect_redshift_serverless(self.engine) - logger.info(f"Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}") + logger.info( + f"Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}" + ) except Exception as exc: - logger.warning(f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}") + logger.warning( + f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}" + ) self.is_serverless = False - + # Set appropriate queries and filters if self.is_serverless: self.sql_stmt = REDSHIFT_SERVERLESS_SQL_STATEMENT diff --git a/ingestion/tests/unit/topology/database/test_redshift_serverless.py b/ingestion/tests/unit/topology/database/test_redshift_serverless.py index 18d33479646c..025453870a19 100644 --- a/ingestion/tests/unit/topology/database/test_redshift_serverless.py +++ b/ingestion/tests/unit/topology/database/test_redshift_serverless.py @@ -38,15 +38,17 @@ def setUp(self): """Set up test fixtures""" self.mock_engine = MagicMock(spec=Engine) self.mock_connection = MagicMock() - self.mock_engine.connect.return_value.__enter__.return_value = self.mock_connection + self.mock_engine.connect.return_value.__enter__.return_value = ( + self.mock_connection + ) def test_detect_redshift_provisioned(self): """Test detection of Redshift Provisioned cluster (STL tables accessible)""" # Mock successful STL query execution self.mock_connection.execute.return_value = None - + result = detect_redshift_serverless(self.mock_engine) - + self.assertFalse(result) self.mock_connection.execute.assert_called_once() @@ -54,13 +56,11 @@ def test_detect_redshift_serverless_insufficient_privilege(self): """Test detection of Redshift Serverless (InsufficientPrivilege error)""" # Mock InsufficientPrivilege error for STL query self.mock_connection.execute.side_effect = ProgrammingError( - "permission denied for relation stl_query", - None, - InsufficientPrivilege() + "permission denied for relation stl_query", None, InsufficientPrivilege() ) - + result = detect_redshift_serverless(self.mock_engine) - + self.assertTrue(result) self.mock_connection.execute.assert_called_once() @@ -68,16 +68,16 @@ def test_detect_redshift_serverless_generic_error(self): """Test detection of Redshift Serverless (generic error)""" # Mock generic error for STL query self.mock_connection.execute.side_effect = Exception("Table does not exist") - + result = detect_redshift_serverless(self.mock_engine) - + self.assertTrue(result) self.mock_connection.execute.assert_called_once() def test_get_redshift_queries_provisioned(self): """Test query selection for Redshift Provisioned""" queries = get_redshift_queries(is_serverless=False) - + self.assertEqual(queries["sql_statement"], REDSHIFT_SQL_STATEMENT) self.assertIn("stl_query", queries["sql_statement"]) self.assertIn("stl_querytext", queries["sql_statement"]) @@ -86,50 +86,60 @@ def test_get_redshift_queries_provisioned(self): def test_get_redshift_queries_serverless(self): """Test query selection for Redshift Serverless""" queries = get_redshift_queries(is_serverless=True) - + self.assertEqual(queries["sql_statement"], REDSHIFT_SERVERLESS_SQL_STATEMENT) self.assertIn("SYS_QUERY_HISTORY", queries["sql_statement"]) self.assertIn("SYS_QUERY_DETAIL", queries["sql_statement"]) self.assertNotIn("stl_query", queries["sql_statement"]) self.assertNotIn("stl_querytext", queries["sql_statement"]) - @patch('metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless') + @patch( + "metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless" + ) def test_usage_source_serverless_initialization(self, mock_detect): """Test RedshiftUsageSource initialization with Serverless detection""" # Mock serverless detection mock_detect.return_value = True - + # Mock config objects - mock_config = MagicMock() - mock_metadata_config = MagicMock() - - with patch.object(RedshiftUsageSource, '__init__', return_value=None) as mock_init: + _ = MagicMock() # config + _ = MagicMock() # metadata_config + + with patch.object( + RedshiftUsageSource, "__init__", return_value=None + ) as _mock_init: # Create instance and manually set up attributes usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) usage_source.engine = self.mock_engine usage_source.is_serverless = True usage_source.sql_stmt = REDSHIFT_SERVERLESS_SQL_STATEMENT - usage_source.filters = usage_source.serverless_filters = """ + usage_source.filters = ( + usage_source.serverless_filters + ) = """ AND query_text NOT ILIKE 'fetch%%' AND query_text NOT ILIKE 'padb_fetch_sample:%%' AND query_text NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' """ - + self.assertTrue(usage_source.is_serverless) self.assertEqual(usage_source.sql_stmt, REDSHIFT_SERVERLESS_SQL_STATEMENT) self.assertIn("query_text", usage_source.filters) - @patch('metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless') + @patch( + "metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless" + ) def test_usage_source_provisioned_initialization(self, mock_detect): """Test RedshiftUsageSource initialization with Provisioned detection""" # Mock provisioned detection mock_detect.return_value = False - + # Mock config objects - mock_config = MagicMock() - mock_metadata_config = MagicMock() - - with patch.object(RedshiftUsageSource, '__init__', return_value=None) as mock_init: + _ = MagicMock() # config + _ = MagicMock() # metadata_config + + with patch.object( + RedshiftUsageSource, "__init__", return_value=None + ) as _mock_init: # Create instance and manually set up attributes usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) usage_source.engine = self.mock_engine @@ -140,7 +150,7 @@ def test_usage_source_provisioned_initialization(self, mock_detect): AND querytxt NOT ILIKE 'padb_fetch_sample:%%' AND querytxt NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' """ - + self.assertFalse(usage_source.is_serverless) self.assertEqual(usage_source.sql_stmt, REDSHIFT_SQL_STATEMENT) self.assertIn("querytxt", usage_source.filters) @@ -148,20 +158,20 @@ def test_usage_source_provisioned_initialization(self, mock_detect): def test_serverless_sql_statement_structure(self): """Test that the serverless SQL statement has the correct structure""" statement = REDSHIFT_SERVERLESS_SQL_STATEMENT - + # Check for SYS views self.assertIn("SYS_QUERY_HISTORY", statement) self.assertIn("SYS_QUERY_DETAIL", statement) - + # Check that STL views are not present self.assertNotIn("stl_query", statement) self.assertNotIn("stl_querytext", statement) self.assertNotIn("stl_scan", statement) - + # Check for proper filtering self.assertIn("status = 'success'", statement) self.assertIn("user_id > 1", statement) - + # Check for placeholder substitution self.assertIn("{start_time}", statement) self.assertIn("{end_time}", statement) @@ -172,25 +182,29 @@ def test_query_factory_returns_correct_types(self): """Test that the query factory returns the expected dictionary structure""" # Test provisioned queries provisioned_queries = get_redshift_queries(is_serverless=False) - expected_keys = ["sql_statement", "test_queries", "table_changes", "metrics_query"] - + expected_keys = [ + "sql_statement", + "test_queries", + "table_changes", + "metrics_query", + ] + for key in expected_keys: self.assertIn(key, provisioned_queries) self.assertIsInstance(provisioned_queries[key], str) - + # Test serverless queries serverless_queries = get_redshift_queries(is_serverless=True) - + for key in expected_keys: self.assertIn(key, serverless_queries) self.assertIsInstance(serverless_queries[key], str) - + # Ensure they're different self.assertNotEqual( - provisioned_queries["sql_statement"], - serverless_queries["sql_statement"] + provisioned_queries["sql_statement"], serverless_queries["sql_statement"] ) if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() From 65f1436394ff1265384905df16deba288ee600a4 Mon Sep 17 00:00:00 2001 From: Mohit Tilala Date: Mon, 22 Dec 2025 14:50:58 +0530 Subject: [PATCH 3/5] Redshift Serverless: SYS view support for lineage, usage, profiler replacing STL/SVL queries --- .../source/database/redshift/connection.py | 93 ++-- .../source/database/redshift/lineage.py | 41 +- .../source/database/redshift/metadata.py | 193 +++++--- .../source/database/redshift/models.py | 8 + .../source/database/redshift/queries.py | 446 +++++++++++------- .../source/database/redshift/usage.py | 37 +- .../source/database/redshift/utils.py | 2 +- .../metrics/system/redshift/system.py | 94 +++- .../unit/topology/database/test_redshift.py | 155 +++++- .../database/test_redshift_serverless.py | 305 +++++++----- .../public/locales/en-US/Database/Redshift.md | 7 +- 11 files changed, 926 insertions(+), 455 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py index 5fb7bcc697ff..828683cd69e9 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py @@ -16,6 +16,7 @@ from typing import Optional from sqlalchemy.engine import Engine +from sqlalchemy.exc import ProgrammingError from sqlalchemy.sql import text from metadata.generated.schema.entity.automations.workflow import ( @@ -41,11 +42,11 @@ ) from metadata.ingestion.ometa.ometa_api import OpenMetadata from metadata.ingestion.source.connections_utils import kill_active_connections +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType from metadata.ingestion.source.database.redshift.queries import ( REDSHIFT_GET_ALL_RELATIONS, REDSHIFT_GET_DATABASE_NAMES, - REDSHIFT_TEST_GET_QUERIES, - REDSHIFT_TEST_GET_SERVERLESS_QUERIES, + REDSHIFT_TEST_GET_QUERIES_MAP, REDSHIFT_TEST_PARTITION_DETAILS, ) from metadata.utils.constants import THREE_MIN @@ -65,28 +66,40 @@ def get_connection(connection: RedshiftConnection) -> Engine: ) -def detect_redshift_serverless(engine: Engine) -> bool: +def get_redshift_instance_type(engine: Engine) -> RedshiftInstanceType: """ - Detect if the Redshift deployment is Serverless or Provisioned. + Detect whether the connected Amazon Redshift deployment is Provisioned + or Serverless by probing for STL system table availability. - Redshift Serverless doesn't have access to STL/SVV system tables, - so we try to query an STL table to determine the deployment type. + Serverless deployments do not have access to STL_* system tables due to + their architecture. Use SYS_* views instead for Serverless compatibility. + + Reference: https://docs.aws.amazon.com/redshift/latest/dg/cm_chap_system-tables.html#sys_view_migration-use_cases Args: - engine: SQLAlchemy engine connected to Redshift + engine (Engine): SQLAlchemy engine connected to a Redshift endpoint. Returns: - bool: True if Serverless, False if Provisioned + RedshiftInstanceType: PROVISIONED if STL tables are accessible, + SERVERLESS otherwise. """ + probe_query = text("SELECT 1 FROM pg_catalog.stl_query LIMIT 1") + try: with engine.connect() as conn: - # Try to access STL_QUERY - this will fail in Serverless - conn.execute(text("SELECT 1 FROM pg_catalog.stl_query LIMIT 1")) - logger.info("Detected Redshift Provisioned cluster (STL tables accessible)") - return False # Provisioned - except Exception as exc: - logger.info(f"Detected Redshift Serverless (STL tables not accessible): {exc}") - return True # Serverless + conn.execute(probe_query) + + logger.info( + "Redshift instance type detected: PROVISIONED (STL tables accessible)" + ) + return RedshiftInstanceType.PROVISIONED + + except ProgrammingError: + logger.info( + "Redshift instance type detected: SERVERLESS " + "(STL tables not accessible, will use SYS_* views)" + ) + return RedshiftInstanceType.SERVERLESS def test_connection( @@ -108,43 +121,39 @@ def test_connection( def test_get_queries_permissions(engine_: Engine): """Check if we have the right permissions to list queries""" - with engine_.connect() as conn: - res = conn.execute(text(REDSHIFT_TEST_GET_QUERIES)).fetchone() - if not all(res): - raise SourceConnectionException( - f"We don't have the right permissions to list queries - {res}" - ) - - def test_get_serverless_queries_permissions(engine_: Engine): - """Check if we have the right permissions to list queries in Serverless""" - with engine_.connect() as conn: - res = conn.execute(text(REDSHIFT_TEST_GET_SERVERLESS_QUERIES)).fetchone() - if not all(res): - raise SourceConnectionException( - f"We don't have the right permissions to list queries in Serverless - {res}" - ) + redshift_instance_type = get_redshift_instance_type(engine_) - # Detect if this is Redshift Serverless - is_serverless = detect_redshift_serverless(engine) + with engine_.connect() as conn: + if redshift_instance_type == RedshiftInstanceType.PROVISIONED: + res = conn.execute( + REDSHIFT_TEST_GET_QUERIES_MAP[RedshiftInstanceType.PROVISIONED] + ).fetchone() + if not all(res): + raise SourceConnectionException( + "We don't have the right permissions to list queries from stl views (Redshift Provisioned)" + f" - {res}" + ) + else: + res = conn.execute( + REDSHIFT_TEST_GET_QUERIES_MAP[RedshiftInstanceType.SERVERLESS] + ).fetchone() + if not all(res): + raise SourceConnectionException( + "We don't have the right permissions to list queries from sys views (Redshift Serverless)" + f" - {res}" + ) test_fn = { "CheckAccess": partial(test_connection_engine_step, engine), "GetSchemas": partial(execute_inspector_func, engine, "get_schema_names"), "GetTables": partial(test_query, statement=table_and_view_query, engine=engine), "GetViews": partial(test_query, statement=table_and_view_query, engine=engine), - "GetQueries": partial( - ( - test_get_serverless_queries_permissions - if is_serverless - else test_get_queries_permissions - ), - engine, - ), + "GetQueries": partial(test_get_queries_permissions, engine), "GetDatabases": partial( - test_query, statement=text(REDSHIFT_GET_DATABASE_NAMES), engine=engine + test_query, statement=REDSHIFT_GET_DATABASE_NAMES, engine=engine ), "GetPartitionTableDetails": partial( - test_query, statement=text(REDSHIFT_TEST_PARTITION_DETAILS), engine=engine + test_query, statement=REDSHIFT_TEST_PARTITION_DETAILS, engine=engine ), } diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/lineage.py b/ingestion/src/metadata/ingestion/source/database/redshift/lineage.py index 9a3acddb24b1..cc9a6803b55b 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/lineage.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/lineage.py @@ -35,12 +35,18 @@ from metadata.generated.schema.type.tableQuery import TableQuery from metadata.ingestion.source.database.lineage_source import LineageSource +from metadata.ingestion.source.database.redshift.connection import ( + get_redshift_instance_type, +) +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType from metadata.ingestion.source.database.redshift.queries import ( - REDSHIFT_GET_STORED_PROCEDURE_QUERIES, - REDSHIFT_SQL_STATEMENT, + REDSHIFT_GET_STORED_PROCEDURE_QUERIES_MAP, + REDSHIFT_SQL_STATEMENT_MAP, ) from metadata.ingestion.source.database.redshift.query_parser import ( + OpenMetadata, RedshiftQueryParserSource, + WorkflowSource, ) from metadata.ingestion.source.database.stored_procedures_mixin import ( StoredProcedureLineageMixin, @@ -54,7 +60,7 @@ class RedshiftLineageSource( RedshiftQueryParserSource, StoredProcedureLineageMixin, LineageSource ): - filters = """ + provisioned_filters = """ AND ( querytxt ILIKE '%%create%%table%%as%%select%%' OR querytxt ILIKE '%%insert%%into%%select%%' @@ -63,7 +69,30 @@ class RedshiftLineageSource( ) """ - sql_stmt = REDSHIFT_SQL_STATEMENT + serverless_filters = """ + AND ( + (query_text ILIKE '%%create%%table%%as%%select%%' AND query_type = 'CTAS') + OR (query_text ILIKE '%%insert%%into%%select%%' AND query_type = 'INSERT') + OR (query_text ILIKE '%%update%%' AND query_type = 'UPDATE') + OR (query_text ILIKE '%%merge%%' AND query_type = 'MERGE') + ) + """ + + def __init__(self, config: WorkflowSource, metadata: OpenMetadata): + super().__init__(config, metadata) + + self.redshift_instance_type = get_redshift_instance_type(self.engine) + + if self.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + self.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.PROVISIONED] + self.filters = self.provisioned_filters + logger.info( + "Using STL views for lineage processing of Redshift Provisioned" + ) + else: + self.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.SERVERLESS] + self.filters = self.serverless_filters + logger.info("Using SYS views for lineage processing of Redshift Serverless") def yield_table_query(self) -> Iterator[TableQuery]: """ @@ -101,6 +130,8 @@ def get_stored_procedure_sql_statement(self) -> str: Return the SQL statement to get the stored procedure queries """ start, _ = get_start_and_end(self.source_config.queryLogDuration) - query = REDSHIFT_GET_STORED_PROCEDURE_QUERIES.format(start_date=start) + query = REDSHIFT_GET_STORED_PROCEDURE_QUERIES_MAP[ + self.redshift_instance_type + ].format(start_date=start) return query diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py b/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py index b9811f355d37..a8f8b76a4288 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/metadata.py @@ -18,7 +18,12 @@ from sqlalchemy import sql from sqlalchemy.dialects.postgresql.base import PGDialect from sqlalchemy.engine.reflection import Inspector -from sqlalchemy_redshift.dialect import RedshiftDialect, RedshiftDialectMixin +from sqlalchemy_redshift.dialect import ( + FOREIGN_KEY_RE, + SQL_IDENTIFIER_RE, + RedshiftDialect, + RedshiftDialectMixin, +) from metadata.generated.schema.api.data.createStoredProcedure import ( CreateStoredProcedureRequest, @@ -66,23 +71,18 @@ LifeCycleQueryMixin, ) from metadata.ingestion.source.database.multi_db_source import MultiDBSource -from metadata.ingestion.source.database.redshift.connection import ( - detect_redshift_serverless, -) from metadata.ingestion.source.database.redshift.incremental_table_processor import ( RedshiftIncrementalTableProcessor, ) from metadata.ingestion.source.database.redshift.models import RedshiftStoredProcedure from metadata.ingestion.source.database.redshift.queries import ( REDSHIFT_EXTERNAL_TABLE_LOCATION, + REDSHIFT_GET_ALL_CONSTRAINTS, REDSHIFT_GET_ALL_RELATION_INFO, REDSHIFT_GET_DATABASE_NAMES, REDSHIFT_GET_STORED_PROCEDURES, REDSHIFT_LIFE_CYCLE_QUERY, REDSHIFT_PARTITION_DETAILS, - REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY, - REDSHIFT_TABLE_CHANGES_QUERY, - get_redshift_queries, ) from metadata.ingestion.source.database.redshift.utils import ( _get_all_relation_info, @@ -100,6 +100,7 @@ calculate_execution_time_generator, ) from metadata.utils.filters import filter_by_database +from metadata.utils.helpers import clean_up_starting_ending_double_quotes_in_string from metadata.utils.logger import ingestion_logger from metadata.utils.sqlalchemy_utils import ( get_all_table_comments, @@ -114,6 +115,7 @@ "r": TableType.Regular, "e": TableType.External, "v": TableType.View, + "m": TableType.MaterializedView, } # pylint: disable=protected-access @@ -148,6 +150,9 @@ def __init__( ): super().__init__(config, metadata) self.partition_details = {} + self.constraint_details: dict[ + str, dict[str, set[str] | list[dict[str, str]]] + ] = {} self.life_cycle_query = REDSHIFT_LIFE_CYCLE_QUERY self.context.get_global().deleted_tables = [] self.incremental = incremental_configuration @@ -156,30 +161,6 @@ def __init__( ] = None self.external_location_map = {} - # Detect Redshift deployment type and set appropriate queries - try: - self.is_serverless = detect_redshift_serverless(self.engine) - logger.info( - f"Detected Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}" - ) - - # Get appropriate queries for the deployment type - self.redshift_queries = get_redshift_queries(self.is_serverless) - - # Update table changes query for incremental processing - if self.is_serverless: - self.table_changes_query = REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY - else: - self.table_changes_query = REDSHIFT_TABLE_CHANGES_QUERY - - except Exception as exc: - logger.warning( - f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}" - ) - self.is_serverless = False - self.redshift_queries = get_redshift_queries(False) - self.table_changes_query = REDSHIFT_TABLE_CHANGES_QUERY - if self.incremental.enabled: logger.info( "Starting Incremental Metadata Extraction.\n\t Considering Table changes from %s", @@ -215,7 +196,9 @@ def get_partition_details(self) -> None: """ try: self.partition_details.clear() - results = self.connection.execute(REDSHIFT_PARTITION_DETAILS).fetchall() + results = self.connection.execute( + statement=REDSHIFT_PARTITION_DETAILS + ).fetchall() for row in results: self.partition_details[f"{row.schema}.{row.table}"] = row.diststyle except Exception as exe: @@ -228,9 +211,18 @@ def query_table_names_and_types( """ Handle custom table types """ + self._set_constraint_details(schema_name) result = self.connection.execute( - sql.text(REDSHIFT_GET_ALL_RELATION_INFO), + sql.text( + REDSHIFT_GET_ALL_RELATION_INFO.format( + view_filter=( + "OR c.relkind IN ('v', 'm')" + if self.source_config.includeViews + else "AND c.relkind NOT IN ('v', 'm')" + ) + ) + ), {"schema": schema_name}, ) @@ -262,23 +254,7 @@ def query_view_names_and_types( This is useful for sources where we need fine-grained logic on how to handle table types, e.g., material views,... """ - - result = self.inspector.get_view_names(schema_name) or [] - - if self.incremental.enabled: - result = [ - name - for name in result - if name - in self.incremental_table_processor.get_not_deleted( - schema_name=schema_name - ) - ] - - return [ - TableNameAndType(name=table_name, type_=TableType.View) - for table_name in result - ] + return [] def get_configured_database(self) -> Optional[str]: if not self.service_connection.ingestAllDatabases: @@ -289,7 +265,7 @@ def get_database_names_raw(self) -> Iterable[str]: yield from self._execute_database_query(REDSHIFT_GET_DATABASE_NAMES) def _set_incremental_table_processor(self, database: str): - """Prepares the needed data for doing incremental metadata extration for a given database. + """Prepares the needed data for doing incremental metadata extraction for a given database. 1. Queries Redshift to get the changes done after the `self.incremental.start_datetime_utc` 2. Sets the table map with the changes within the RedshiftIncrementalTableProcessor @@ -327,7 +303,6 @@ def set_external_location_map(self, database_name: str) -> None: } def get_database_names(self) -> Iterable[str]: - """Get database names to process.""" if not self.config.serviceConnection.root.config.ingestAllDatabases: configured_db = self.config.serviceConnection.root.config.database self.get_partition_details() @@ -378,7 +353,7 @@ def _get_partition_key(self, diststyle: str) -> Optional[str]: @calculate_execution_time() def get_table_partition_details( - self, table_name: str, schema_name: str, _inspector: Inspector + self, table_name: str, schema_name: str, inspector: Inspector ) -> Tuple[bool, Optional[TablePartition]]: diststyle = self.partition_details.get(f"{schema_name}.{table_name}") if diststyle: @@ -487,3 +462,115 @@ def mark_tables_as_deleted(self): ) else: yield from super().mark_tables_as_deleted() + + def _get_columns_with_constraints( + self, + schema_name: str, + table_name: str, + *args, + **kwargs, + ) -> tuple[list[str], list[str], list[str]]: + """Fetch constraint for a specific schema and table + + Args: + schema_name (str): schema name + table_name (str): table name + + Returns: + tuple[list, list, list]: list of primary, unique and foreign columns + """ + constraints = self.constraint_details.get(f"{schema_name}.{table_name}", {}) + if not constraints: + return [], [], [] + pkeys = [ + clean_up_starting_ending_double_quotes_in_string(p) + for p in constraints.get("pkey", set()) + ] + ukeys = [ + clean_up_starting_ending_double_quotes_in_string(p) + for p in constraints.get("ukey", set()) + ] + + fkeys = [] + fkey_constraints: list[dict[str, str]] = constraints.get("fkey", []) + for fkey_constraint in fkey_constraints: + fkey_constraint.update( + { + "constrained_columns": [ + clean_up_starting_ending_double_quotes_in_string(column) + for column in fkey_constraint.get("constrained_columns") + ], + "referred_columns": [ + clean_up_starting_ending_double_quotes_in_string(column) + for column in fkey_constraint.get("referred_columns") + ], + } + ) + fkeys.append(fkey_constraint) + + return pkeys, [ukeys], fkeys + + def _set_constraint_details(self, schema_name: str): + """Get all the column constraints in a given schema + + Args: + schema_name (str): schema name + """ + self.constraint_details = ( + {} + ) # reset constraint_details dict when fetching for a new schema + + rows = self.connection.execute( + sql.text(REDSHIFT_GET_ALL_CONSTRAINTS), + {"schema": schema_name}, + ) + + for row in rows or []: + schema_table_name = f"{row.schema}.{row.table_name}" + schema_table_constraints = self.constraint_details.setdefault( + schema_table_name, {} + ) + if row.constraint_type == "p": + pkey = schema_table_constraints.setdefault("pkey", set()) + pkey.add(row.column_name) + if row.constraint_type == "f": + fkey_constraint = { + "key": row.conkey, + "condef": row.condef, + "database": self.connection.engine.url.database, + } + extracted_fkey = self._extract_fkeys(fkey_constraint) + fkey: list[dict[str, str]] = schema_table_constraints.setdefault( + "fkey", [] + ) + fkey.extend(extracted_fkey) + if row.constraint_type == "u": + ukey = schema_table_constraints.setdefault("ukey", set()) + ukey.add(row.column_name) + + def _extract_fkeys(self, fkey_constraint: dict) -> list[dict[str, str]]: + """extract foreign keys from rows + + Args: + uniques (dict): _description_ + """ + fkeys = [] + + m = FOREIGN_KEY_RE.match(fkey_constraint["condef"]) + colstring = m.group("referred_columns") + referred_columns = SQL_IDENTIFIER_RE.findall(colstring) + referred_table = m.group("referred_table") + referred_schema = m.group("referred_schema") + colstring = m.group("columns") + constrained_columns = SQL_IDENTIFIER_RE.findall(colstring) + fkey_d = { + "name": fkey_constraint["key"], + "constrained_columns": constrained_columns, + "referred_schema": referred_schema, + "referred_table": referred_table, + "referred_columns": referred_columns, + "referred_database": fkey_constraint["database"], + } + fkeys.append(fkey_d) + + return fkeys diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/models.py b/ingestion/src/metadata/ingestion/source/database/redshift/models.py index 3423b8afa3a6..2ea274dc261a 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/models.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/models.py @@ -12,6 +12,7 @@ Redshift models """ import re +from enum import Enum from typing import Dict, List, Optional, Tuple from pydantic import BaseModel @@ -20,6 +21,13 @@ SchemaName = str +class RedshiftInstanceType(Enum): + """Redshift Instance Types""" + + PROVISIONED = "PROVISIONED" + SERVERLESS = "SERVERLESS" + + class RedshiftStoredProcedure(BaseModel): """Redshift stored procedure list query results""" diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py index bcd0e410c400..17f555b6c851 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py @@ -13,91 +13,8 @@ """ import textwrap -from typing import List -from metadata.utils.profiler_utils import QueryResult -from metadata.utils.time_utils import datetime_to_timestamp - - -def get_redshift_queries(is_serverless: bool = False) -> dict: - """ - Get the appropriate set of queries based on Redshift deployment type. - - Args: - is_serverless: True for Serverless, False for Provisioned - - Returns: - dict: Dictionary mapping query names to query strings - """ - if is_serverless: - return { - "sql_statement": REDSHIFT_SERVERLESS_SQL_STATEMENT, - "test_queries": REDSHIFT_TEST_GET_SERVERLESS_QUERIES, - "table_changes": REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY, - "metrics_query": SERVERLESS_QUERY_METRICS, - } - return { - "sql_statement": REDSHIFT_SQL_STATEMENT, - "test_queries": REDSHIFT_TEST_GET_QUERIES, - "table_changes": REDSHIFT_TABLE_CHANGES_QUERY, - "metrics_query": STL_QUERY, - } - - -# Serverless-compatible query using SYS views -REDSHIFT_SERVERLESS_SQL_STATEMENT = textwrap.dedent( - """ - WITH - queries AS ( - SELECT - query_id, - user_id, - user_name, - database_name, - query_text, - start_time, - end_time, - status - FROM SYS_QUERY_HISTORY - WHERE user_id > 1 - {filters} - -- Filter out all automated & cursor queries - AND query_text NOT LIKE '/* {{"app": "OpenMetadata", %%}} */%%' - AND query_text NOT LIKE '/* {{"app": "dbt", %%}} */%%' - AND user_id <> 1 - AND status = 'success' - AND start_time >= '{start_time}' - AND end_time < '{end_time}' - LIMIT {result_limit} - ), - table_access AS ( - -- Get table access information from query details - SELECT DISTINCT - qh.query_id, - qd.table_id, - sti.database AS database_name, - sti.schema AS schema_name - FROM queries qh - INNER JOIN SYS_QUERY_DETAIL qd ON qh.query_id = qd.query_id - INNER JOIN pg_catalog.svv_table_info sti ON qd.table_id = sti.table_id - WHERE qd.table_id IS NOT NULL - ) - SELECT DISTINCT - q.user_id, - q.query_id, - RTRIM(q.user_name) AS user_name, - q.query_text, - ta.database_name, - ta.schema_name, - q.start_time, - q.end_time, - DATEDIFF(millisecond, q.start_time, q.end_time) AS duration, - CASE WHEN q.status = 'success' THEN 0 ELSE 1 END AS aborted - FROM queries AS q - LEFT JOIN table_access AS ta ON ta.query_id = q.query_id - ORDER BY q.end_time DESC -""" -) +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType # Not able to use SYS_QUERY_HISTORY here. Few users not getting any results REDSHIFT_SQL_STATEMENT = textwrap.dedent( @@ -112,7 +29,6 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: AND label NOT IN ('maintenance', 'metrics', 'health') AND querytxt NOT LIKE '/* {{"app": "OpenMetadata", %%}} */%%' AND querytxt NOT LIKE '/* {{"app": "dbt", %%}} */%%' - AND userid <> 1 AND aborted = 0 AND starttime >= '{start_time}' AND starttime < '{end_time}' @@ -131,7 +47,7 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: LISTAGG(CASE WHEN LEN(RTRIM(text)) = 0 THEN text ELSE RTRIM(text) END, '') WITHIN GROUP (ORDER BY sequence) AS query_text FROM deduped_querytext - WHERE sequence < 327 -- each chunk contains up to 200, RS has a maximum str length of 65535. + WHERE sequence < 327 -- each chunk contains up to 200, RS has a maximum str length of 65535. GROUP BY query ), raw_scans AS ( @@ -140,11 +56,11 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: FROM pg_catalog.stl_scan ), scans AS ( - SELECT DISTINCT - query, - sti.database AS database_name, + SELECT DISTINCT + query, + sti.database AS database_name, sti.schema AS schema_name - FROM raw_scans AS s + FROM raw_scans AS s INNER JOIN pg_catalog.svv_table_info AS sti ON (s.tbl)::oid = sti.table_id ) @@ -171,6 +87,75 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: ) +REDSHIFT_SERVERLESS_SQL_STATEMENT = textwrap.dedent( + """ +WITH queries AS ( + SELECT * + FROM SYS_QUERY_HISTORY + WHERE user_id > 1 + {filters} + -- Filter out all automated & cursor queries + AND LOWER(query_label) NOT IN ('maintenance', 'metrics', 'health') + AND query_text NOT LIKE '/* {{"app": "OpenMetadata", %%}} */%%' + AND query_text NOT LIKE '/* {{"app": "dbt", %%}} */%%' + AND LOWER(status) = 'success' + AND start_time >= '{start_time}' + AND start_time < '{end_time}' +), +deduped_querytext AS ( + -- Sometimes rows are duplicated, causing LISTAGG to fail in the full_queries CTE. + SELECT DISTINCT + qt.* + FROM SYS_QUERY_TEXT AS qt + INNER JOIN queries AS q + ON qt.query_id = q.query_id +), +full_queries AS ( + SELECT + query_id, + LISTAGG(CASE WHEN LEN(RTRIM(text)) = 0 THEN text ELSE RTRIM(text) END, '') + WITHIN GROUP (ORDER BY sequence) AS query_text + FROM deduped_querytext + WHERE sequence < 327 -- each chunk contains up to 200, RS has a maximum str length of 65535. + GROUP BY query_id +), +query_detail AS ( + SELECT DISTINCT + query_id, + COALESCE(SPLIT_PART(table_name, '.', 1), 'unknown') AS database_name, + COALESCE(SPLIT_PART(table_name, '.', 2), 'unknown') AS schema_name, + COALESCE(SPLIT_PART(table_name, '.', 3), table_name) AS table_name + FROM SYS_QUERY_DETAIL + WHERE LOWER(step_name) IN ('insert', 'merge', 'delete') + AND table_name <> '' +) +SELECT DISTINCT + q.user_id, + q.query_id, + RTRIM(u.usename) AS user_name, + fq.query_text, + qd.database_name, + qd.schema_name, + q.start_time AS start_time, + q.end_time AS end_time, + datediff(millisecond, q.start_time, q.end_time) AS duration, + q.status, + q.query_type +FROM queries AS q + -- instances where query_detail has no entries of step 'insert', 'merge' or 'delete' + -- means no table was affected so perform inner join to filter those out + INNER JOIN query_detail AS qd + ON q.query_id = qd.query_id + INNER JOIN full_queries AS fq + ON q.query_id = fq.query_id + INNER JOIN pg_catalog.pg_user AS u + ON q.user_id = u.usesysid +ORDER BY q.start_time DESC +LIMIT {result_limit} +""" +) + + REDSHIFT_GET_ALL_RELATION_INFO = textwrap.dedent( """ SELECT @@ -178,7 +163,7 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: c.relkind FROM pg_catalog.pg_class c LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace - WHERE c.relkind = 'r' + WHERE (c.relkind IN ('r', 'S', 'f') {view_filter}) AND n.nspname = :schema UNION SELECT @@ -312,22 +297,57 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: """ REDSHIFT_TEST_GET_QUERIES = """ -SELECT - has_table_privilege('svv_table_info', 'SELECT') as can_access_svv_table_info, - has_table_privilege('stl_querytext', 'SELECT') as can_access_stl_querytext, - has_table_privilege('stl_query', 'SELECT') as can_access_stl_query; +SELECT + has_table_privilege('SVV_TABLE_INFO', 'SELECT') as can_access_svv_table_info, + has_table_privilege('STL_QUERY', 'SELECT') as can_access_stl_query, + has_table_privilege('STL_QUERYTEXT', 'SELECT') as can_access_stl_querytext, + has_table_privilege('STL_SCAN', 'SELECT') as can_access_stl_scan, + has_table_privilege('SVL_STORED_PROC_CALL', 'SELECT') as can_access_stl_stored_proc_call, + has_table_privilege('STL_INSERT', 'SELECT') as can_access_stl_insert, + has_table_privilege('STL_DELETE', 'SELECT') as can_access_stl_delete; """ -REDSHIFT_TEST_GET_SERVERLESS_QUERIES = """ -SELECT - has_table_privilege('svv_table_info', 'SELECT') as can_access_svv_table_info, +REDSHIFT_SERVERLESS_TEST_GET_QUERIES = """ +SELECT + has_table_privilege('SVV_TABLE_INFO', 'SELECT') as can_access_svv_table_info, + has_table_privilege('SYS_QUERY_HISTORY', 'SELECT') as can_access_sys_query_history, has_table_privilege('SYS_QUERY_TEXT', 'SELECT') as can_access_sys_query_text, - has_table_privilege('SYS_QUERY_HISTORY', 'SELECT') as can_access_sys_query_history; + has_table_privilege('SYS_QUERY_DETAIL', 'SELECT') as can_access_sys_query_detail, + has_table_privilege('SYS_PROCEDURE_CALL', 'SELECT') as can_access_sys_procedure_call; """ REDSHIFT_TEST_PARTITION_DETAILS = "select * from SVV_TABLE_INFO limit 1" +REDSHIFT_GET_ALL_CONSTRAINTS = """ +select + n.nspname as "schema", + c.relname as "table_name", + t.contype as "constraint_type", + t.conkey, + pg_catalog.pg_get_constraintdef(t.oid, true)::varchar(512) as condef, + a.attname as "column_name" +FROM pg_catalog.pg_class c +LEFT JOIN pg_catalog.pg_namespace n + ON n.oid = c.relnamespace +JOIN pg_catalog.pg_constraint t + ON t.conrelid = c.oid +JOIN pg_catalog.pg_attribute a + ON t.conrelid = a.attrelid AND a.attnum = ANY(t.conkey) +WHERE n.nspname not like '^pg_' and schema=:schema +UNION +SELECT + s.schemaname AS "schema", + c.tablename AS "table_name", + 'p' as "constraint_type", + null as conkey, + null as condef, + c.columnname as "column_name" +FROM + svv_external_columns c + JOIN svv_external_schemas s ON s.schemaname = c.schemaname +where 1 and schema=:schema; +""" # Redshift views definitions only contains the select query # hence we are appending "create view .
as " to select query @@ -448,6 +468,66 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: """ ) +REDSHIFT_SERVERLESS_GET_STORED_PROCEDURE_QUERIES = textwrap.dedent( + """ +WITH SP_HISTORY AS ( + SELECT + spcall.query_text AS procedure_text, + spcall.start_time AS procedure_start_time, + spcall.end_time AS procedure_end_time, + spcall.query_id AS procedure_query_id, + qh.session_id AS procedure_session_id + FROM SYS_PROCEDURE_CALL spcall + LEFT JOIN SYS_QUERY_HISTORY qh + ON spcall.query_id = qh.query_id + WHERE LOWER(spcall.status) = 'success' + AND spcall.start_time >= '{start_date}' +), +Q_HISTORY AS ( + SELECT + query_text, + CASE + WHEN query_text ILIKE '%%MERGE%%' THEN 'MERGE' + WHEN query_text ILIKE '%%UPDATE%%' THEN 'UPDATE' + WHEN query_text ILIKE '%%CREATE%%AS%%' THEN 'CREATE_TABLE_AS_SELECT' + WHEN query_text ILIKE '%%INSERT%%' THEN 'INSERT' + ELSE 'UNKNOWN' END query_type, + database_name AS query_database_name, + session_id as query_session_id, + start_time AS query_start_time, + end_time AS query_end_time, + CAST(b.usename AS varchar) AS query_user_name + FROM SYS_QUERY_HISTORY q + JOIN pg_catalog.pg_user b + ON b.usesysid = q.user_id + WHERE LOWER(q.query_label) NOT IN ('maintenance', 'metrics', 'health') + AND q.query_text NOT LIKE '/* {{"app": "OpenMetadata", %%}} */%%' + AND q.query_text NOT LIKE '/* {{"app": "dbt", %%}} */%%' + AND LOWER(q.status) = 'success' + AND q.start_time >= '{start_date}' + AND q.user_id <> 1 +) +SELECT + TRIM(sp.procedure_text) procedure_text, + sp.procedure_start_time, + sp.procedure_end_time, + TRIM(q.query_text) query_text, + q.query_type, + TRIM(q.query_database_name) query_database_name, + null AS query_schema_name, + q.query_start_time, + q.query_end_time, + TRIM(q.query_user_name) query_user_name +FROM SP_HISTORY sp + JOIN Q_HISTORY q + ON sp.procedure_session_id = q.query_session_id + AND q.query_start_time BETWEEN sp.procedure_start_time AND sp.procedure_end_time + AND q.query_end_time BETWEEN sp.procedure_start_time AND sp.procedure_end_time +ORDER BY procedure_start_time DESC + """ +) + + REDSHIFT_LIFE_CYCLE_QUERY = textwrap.dedent( """ select "table" as table_name, @@ -473,79 +553,107 @@ def get_redshift_queries(is_serverless: bool = False) -> dict: ORDER BY end_time DESC """ -# Serverless version of table changes query (same as above since it already uses SYS views) -REDSHIFT_SERVERLESS_TABLE_CHANGES_QUERY = REDSHIFT_TABLE_CHANGES_QUERY - - -STL_QUERY = """ - with data as ( - select - {alias}.* - from - pg_catalog.stl_insert si - {join_type} join pg_catalog.stl_delete sd on si.query = sd.query - where - {condition} - ) - SELECT - SUM(data."rows") AS "rows", - sti."database", - sti."schema", - sti."table", - DATE_TRUNC('second', data.starttime) AS starttime - FROM - data - INNER JOIN pg_catalog.svv_table_info sti ON data.tbl = sti.table_id + +REDSHIFT_SYSTEM_METRICS_QUERY = """ +with data as ( + select + {alias}.* + from + pg_catalog.stl_insert si + {join_type} join pg_catalog.stl_delete sd on si.query = sd.query where - sti."database" = '{database}' AND - sti."schema" = '{schema}' AND - "rows" != 0 AND - DATE(data.starttime) >= CURRENT_DATE - 1 - GROUP BY 2,3,4,5 - ORDER BY 5 DESC + {condition} +) +SELECT + SUM(data."rows") AS "rows", + sti."database", + sti."schema", + sti."table", + DATE_TRUNC('second', data.starttime) AS starttime +FROM + data + INNER JOIN pg_catalog.svv_table_info sti ON data.tbl = sti.table_id +where + sti."database" = '{database}' AND + sti."schema" = '{schema}' AND + "rows" != 0 AND + DATE(data.starttime) >= CURRENT_DATE - 1 +GROUP BY 2,3,4,5 +ORDER BY 5 DESC """ -# Serverless-compatible query using SYS views for metrics -SERVERLESS_QUERY_METRICS = """ +# output_rows from SYS_QUERY_DETAIL should match rows from stl_insert/stl_delete +# It’s often wrong (usually too high). I noticed that sometimes the 'scan' step +# with plan_parent_id > 0 and plan_node_id > 0 gives the correct count, taking the +# min id if there are multiple scans. It worked in all the cases I tried, but it’s +# not really reliable for general use. +# For now, we just use the number of queries as a placeholder until we figure out +# a proper fix. +REDSHIFT_SERVERLESS_SYSTEM_METRICS_QUERY = """ +WITH data AS ( SELECT - COUNT(*) AS "rows", - qh.database_name AS "database", - COALESCE( - REGEXP_SUBSTR(qh.query_text, 'FROM\\s+(\\w+)\\.(\\w+)', 1, 1, 'i', 2), - 'unknown' - ) AS "schema", - COALESCE( - REGEXP_SUBSTR(qh.query_text, 'FROM\\s+(\\w+)\\.(\\w+)\\.(\\w+)', 1, 1, 'i', 3), - REGEXP_SUBSTR(qh.query_text, 'FROM\\s+(\\w+)', 1, 1, 'i', 1), - 'unknown' - ) AS "table", - DATE_TRUNC('second', qh.start_time) AS starttime - FROM SYS_QUERY_HISTORY qh - WHERE qh.database_name = '{database}' - AND qh.status = 'success' - AND (qh.query_type = 'INSERT' OR qh.query_type = 'DELETE' OR qh.query_type = 'UPDATE') - AND DATE(qh.start_time) >= CURRENT_DATE - 1 - GROUP BY 2, 3, 4, 5 - ORDER BY 5 DESC + {alias}.output_rows AS rows, + COALESCE(SPLIT_PART({alias}.table_name, '.', 1), 'unknown') AS database, + COALESCE(SPLIT_PART({alias}.table_name, '.', 2), 'unknown') AS schema, + COALESCE(SPLIT_PART({alias}.table_name, '.', 3), {alias}.table_name) AS table, + DATE_TRUNC('second', {alias}.start_time) AS starttime + FROM ( + SELECT * + FROM SYS_QUERY_DETAIL + WHERE lower(step_name) = 'insert' + ) si + {join_type} JOIN ( + SELECT * + FROM SYS_QUERY_DETAIL + WHERE lower(step_name) = 'delete' + ) sd + ON si.query_id = sd.query_id + WHERE + {condition} + AND {alias}.table_name <> '' + AND DATE({alias}.start_time) >= CURRENT_DATE - 1 + AND {alias}.output_rows <> 0 +) +SELECT + COUNT(data.rows) as rows, + data.database, + data.schema, + data.table, + data.starttime +FROM data +WHERE + lower(data.database) = lower('{database}') + AND lower(data.schema) = lower('{schema}') +GROUP BY + data.database, + data.schema, + data.table, + data.starttime +ORDER BY data.starttime DESC; """ - -def get_metric_result(ddls: List[QueryResult], table_name: str) -> List: - """Given query results, retur the metric result - - Args: - ddls (List[QueryResult]): list of query results - table_name (str): table name - - Returns: - List: - """ - return [ - { - "timestamp": datetime_to_timestamp(ddl.start_time, milliseconds=True), - "operation": ddl.query_type, - "rowsAffected": ddl.rows, - } - for ddl in ddls - if ddl.table_name == table_name - ] +# Ideally, all serverless specific queries defined here should work with +# both Redshift Serverless and Provisioned since sys views are available +# in both instances. However, it still needs to be tested in Provisioned +# clusters. +# Ref: https://github.com/open-metadata/OpenMetadata/pull/6568/files#diff-65e5e8591345679be6a347ea29c4d283d5ca9aa723ef788c9a2524344de49ff3R17 + +REDSHIFT_TEST_GET_QUERIES_MAP = { + RedshiftInstanceType.PROVISIONED: REDSHIFT_TEST_GET_QUERIES, + RedshiftInstanceType.SERVERLESS: REDSHIFT_SERVERLESS_TEST_GET_QUERIES, +} + +REDSHIFT_SQL_STATEMENT_MAP = { + RedshiftInstanceType.PROVISIONED: REDSHIFT_SQL_STATEMENT, + RedshiftInstanceType.SERVERLESS: REDSHIFT_SERVERLESS_SQL_STATEMENT, +} + +REDSHIFT_GET_STORED_PROCEDURE_QUERIES_MAP = { + RedshiftInstanceType.PROVISIONED: REDSHIFT_GET_STORED_PROCEDURE_QUERIES, + RedshiftInstanceType.SERVERLESS: REDSHIFT_SERVERLESS_GET_STORED_PROCEDURE_QUERIES, +} + +REDSHIFT_SYSTEM_METRICS_QUERY_MAP = { + RedshiftInstanceType.PROVISIONED: REDSHIFT_SYSTEM_METRICS_QUERY, + RedshiftInstanceType.SERVERLESS: REDSHIFT_SERVERLESS_SYSTEM_METRICS_QUERY, +} diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/usage.py b/ingestion/src/metadata/ingestion/source/database/redshift/usage.py index ff252668e373..1b85dadf9d35 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/usage.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/usage.py @@ -12,11 +12,11 @@ Redshift usage module """ from metadata.ingestion.source.database.redshift.connection import ( - detect_redshift_serverless, + get_redshift_instance_type, ) +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType from metadata.ingestion.source.database.redshift.queries import ( - REDSHIFT_SERVERLESS_SQL_STATEMENT, - REDSHIFT_SQL_STATEMENT, + REDSHIFT_SQL_STATEMENT_MAP, ) from metadata.ingestion.source.database.redshift.query_parser import ( RedshiftQueryParserSource, @@ -30,13 +30,13 @@ class RedshiftUsageSource(RedshiftQueryParserSource, UsageSource): """Redshift Usage Source with support for both Provisioned and Serverless deployments.""" - filters = """ + provisioned_filters = """ AND querytxt NOT ILIKE 'fetch%%' AND querytxt NOT ILIKE 'padb_fetch_sample:%%' AND querytxt NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' """ - # Serverless uses different filter syntax for query_text vs querytxt + # Serverless uses SYS views instead of STL views and have query_text column instead of querytxt serverless_filters = """ AND query_text NOT ILIKE 'fetch%%' AND query_text NOT ILIKE 'padb_fetch_sample:%%' @@ -46,24 +46,13 @@ class RedshiftUsageSource(RedshiftQueryParserSource, UsageSource): def __init__(self, config, metadata_config): super().__init__(config, metadata_config) - # Detect Redshift deployment type - try: - self.is_serverless = detect_redshift_serverless(self.engine) - logger.info( - f"Redshift deployment type: {'Serverless' if self.is_serverless else 'Provisioned'}" - ) - except Exception as exc: - logger.warning( - f"Could not detect Redshift deployment type, defaulting to Provisioned: {exc}" - ) - self.is_serverless = False + self.redshift_instance_type = get_redshift_instance_type(self.engine) - # Set appropriate queries and filters - if self.is_serverless: - self.sql_stmt = REDSHIFT_SERVERLESS_SQL_STATEMENT - self.filters = self.serverless_filters - logger.info("Using SYS views for Redshift Serverless") + if self.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + self.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.PROVISIONED] + self.filters = self.provisioned_filters + logger.info("Using STL views for usage processing of Redshift Provisioned") else: - self.sql_stmt = REDSHIFT_SQL_STATEMENT - self.filters = self.filters - logger.info("Using STL views for Redshift Provisioned") + self.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.SERVERLESS] + self.filters = self.serverless_filters + logger.info("Using SYS views for usage processing of Redshift Serverless") diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/utils.py b/ingestion/src/metadata/ingestion/source/database/redshift/utils.py index 0eeb1dff12bc..5531a3f32158 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/utils.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/utils.py @@ -109,7 +109,7 @@ def _get_column_info(self, *args, **kwargs): )._get_column_info(*args, **kwdrs) # raw_data_type is not included in column_info as - # redhift doesn't support complex data types directly + # redshift doesn't support complex data types directly # https://docs.aws.amazon.com/redshift/latest/dg/c_Supported_data_types.html if "info" not in column_info: diff --git a/ingestion/src/metadata/profiler/metrics/system/redshift/system.py b/ingestion/src/metadata/profiler/metrics/system/redshift/system.py index 376d28ae49ad..c6506dc54980 100644 --- a/ingestion/src/metadata/profiler/metrics/system/redshift/system.py +++ b/ingestion/src/metadata/profiler/metrics/system/redshift/system.py @@ -8,7 +8,13 @@ from sqlalchemy.orm import Session from metadata.generated.schema.entity.data.table import SystemProfile -from metadata.ingestion.source.database.redshift.queries import STL_QUERY +from metadata.ingestion.source.database.redshift.connection import ( + get_redshift_instance_type, +) +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType +from metadata.ingestion.source.database.redshift.queries import ( + REDSHIFT_SYSTEM_METRICS_QUERY_MAP, +) from metadata.profiler.metrics.system.dml_operation import DatabaseDMLOperations from metadata.profiler.metrics.system.system import ( CacheProvider, @@ -37,6 +43,9 @@ def __init__( self.table = runner.table_name self.database = runner.session.get_bind().url.database self.schema = runner.schema_name + self.engine = runner.session.get_bind() + + self.redshift_instance_type = get_redshift_instance_type(self.engine) def get_inserts(self) -> List[SystemProfile]: queries = self.get_or_update_cache( @@ -66,13 +75,26 @@ def get_updates(self) -> List[SystemProfile]: return get_metric_result(queries, self.table) def _get_insert_queries(self, database: str, schema: str) -> List[QueryResult]: - insert_query = STL_QUERY.format( - alias="si", - join_type="LEFT", - condition="sd.query is null", - database=database, - schema=schema, - ) + if self.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + insert_query = REDSHIFT_SYSTEM_METRICS_QUERY_MAP[ + RedshiftInstanceType.PROVISIONED + ].format( + alias="si", + join_type="LEFT", + condition="sd.query is null", + database=database, + schema=schema, + ) + else: + insert_query = REDSHIFT_SYSTEM_METRICS_QUERY_MAP[ + RedshiftInstanceType.SERVERLESS + ].format( + alias="si", + join_type="LEFT", + condition="sd.query_id is null", + database=database, + schema=schema, + ) return self._get_query_results( self.session, insert_query, @@ -80,13 +102,26 @@ def _get_insert_queries(self, database: str, schema: str) -> List[QueryResult]: ) def _get_delete_queries(self, database: str, schema: str) -> List[QueryResult]: - delete_query = STL_QUERY.format( - alias="sd", - join_type="RIGHT", - condition="si.query is null", - database=database, - schema=schema, - ) + if self.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + delete_query = REDSHIFT_SYSTEM_METRICS_QUERY_MAP[ + RedshiftInstanceType.PROVISIONED + ].format( + alias="sd", + join_type="RIGHT", + condition="si.query is null", + database=database, + schema=schema, + ) + else: + delete_query = REDSHIFT_SYSTEM_METRICS_QUERY_MAP[ + RedshiftInstanceType.SERVERLESS + ].format( + alias="sd", + join_type="RIGHT", + condition="si.query_id is null", + database=database, + schema=schema, + ) return self._get_query_results( self.session, delete_query, @@ -94,13 +129,26 @@ def _get_delete_queries(self, database: str, schema: str) -> List[QueryResult]: ) def _get_update_queries(self, database: str, schema: str) -> List[QueryResult]: - update_query = STL_QUERY.format( - alias="si", - join_type="INNER", - condition="sd.query is not null", - database=database, - schema=schema, - ) + if self.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + update_query = REDSHIFT_SYSTEM_METRICS_QUERY_MAP[ + RedshiftInstanceType.PROVISIONED + ].format( + alias="si", + join_type="INNER", + condition="sd.query is not null", + database=database, + schema=schema, + ) + else: + update_query = REDSHIFT_SYSTEM_METRICS_QUERY_MAP[ + RedshiftInstanceType.SERVERLESS + ].format( + alias="si", + join_type="INNER", + condition="sd.query_id is not null", + database=database, + schema=schema, + ) return self._get_query_results( self.session, update_query, @@ -109,7 +157,7 @@ def _get_update_queries(self, database: str, schema: str) -> List[QueryResult]: def get_metric_result(ddls: List[QueryResult], table_name: str) -> List[SystemProfile]: - """Given query results, retur the metric result + """Given query results, return the metric result Args: ddls (List[QueryResult]): list of query results diff --git a/ingestion/tests/unit/topology/database/test_redshift.py b/ingestion/tests/unit/topology/database/test_redshift.py index bc2d5128518b..b22bff9a9dea 100644 --- a/ingestion/tests/unit/topology/database/test_redshift.py +++ b/ingestion/tests/unit/topology/database/test_redshift.py @@ -10,16 +10,23 @@ # limitations under the License. """ -Test Redshift using the topology +Test Redshift Provisioned cluster detection and query selection """ -from unittest import TestCase -from unittest.mock import patch +import unittest +from unittest.mock import Mock, patch from metadata.generated.schema.metadataIngestion.workflow import ( OpenMetadataWorkflowConfig, ) +from metadata.ingestion.source.database.redshift.connection import ( + get_redshift_instance_type, +) from metadata.ingestion.source.database.redshift.metadata import RedshiftSource +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType +from metadata.ingestion.source.database.redshift.queries import ( + REDSHIFT_SQL_STATEMENT_MAP, +) mock_redshift_config = { "source": { @@ -56,13 +63,15 @@ EXPECTED_PARTITION_COLUMNS = ["eventid", None, None] -class RedshiftUnitTest(TestCase): +class RedshiftUnitTest(unittest.TestCase): + """Test cases for Redshift Provisioned cluster""" + @patch( "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.test_connection" ) - def __init__(self, methodName, test_connection) -> None: - super().__init__(methodName) - test_connection.return_value = False + def setUp(self, mock_test_connection): + """Set up test fixtures""" + mock_test_connection.return_value = False self.config = OpenMetadataWorkflowConfig.model_validate(mock_redshift_config) self.redshift_source = RedshiftSource.create( mock_redshift_config["source"], @@ -70,6 +79,7 @@ def __init__(self, methodName, test_connection) -> None: ) def test_partition_parse_columns(self): + """Test parsing of partition key from distribution style""" for i in range(len(RAW_DIST_STYLE)): with self.subTest(i=i): self.assertEqual( @@ -81,6 +91,133 @@ def test_partition_parse_columns(self): @patch( "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.connection" ) - def test_close_connection(self, engine, connection): - connection.return_value = True + def test_close_connection(self, mock_connection, mock_engine): + """Test connection closing""" + mock_connection.return_value = True self.redshift_source.close() + + def test_detect_provisioned_when_stl_accessible(self): + """Test detection of Provisioned cluster when STL tables are accessible""" + mock_engine = Mock() + mock_conn = Mock() + mock_context = Mock() + mock_context.__enter__ = Mock(return_value=mock_conn) + mock_context.__exit__ = Mock(return_value=False) + mock_engine.connect.return_value = mock_context + mock_conn.execute.return_value = Mock() # STL query succeeds + + result = get_redshift_instance_type(mock_engine) + + self.assertEqual(result, RedshiftInstanceType.PROVISIONED) + mock_conn.execute.assert_called_once() + + def test_provisioned_uses_stl_queries(self): + """Test that Provisioned cluster uses STL-based queries""" + sql = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.PROVISIONED] + + # Check for use of STL tables + self.assertIn("stl_query", sql.lower()) + self.assertIn("stl_querytext", sql.lower()) + + # Check that SYS views are not used + self.assertNotIn("SYS_QUERY_HISTORY", sql) + self.assertNotIn("SYS_QUERY_TEXT", sql) + self.assertNotIn("SYS_QUERY_DETAIL", sql) + + # Check for proper placeholder substitution + self.assertIn("{start_time}", sql) + self.assertIn("{end_time}", sql) + self.assertIn("{result_limit}", sql) + self.assertIn("{filters}", sql) + + @patch( + "metadata.ingestion.source.database.redshift.usage.get_redshift_instance_type" + ) + def test_usage_source_provisioned_initialization(self, mock_get_instance_type): + """Test RedshiftUsageSource filters and SQL statement for Provisioned""" + from metadata.ingestion.source.database.redshift.usage import ( + RedshiftUsageSource, + ) + + mock_get_instance_type.return_value = RedshiftInstanceType.PROVISIONED + mock_engine = Mock() + + # Create instance without full initialization + usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) + usage_source.engine = mock_engine + usage_source.redshift_instance_type = mock_get_instance_type.return_value + + # Simulate __init__ logic for filter and statement selection + if usage_source.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + usage_source.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[ + RedshiftInstanceType.PROVISIONED + ] + usage_source.filters = RedshiftUsageSource.provisioned_filters + + # Verify instance type detected correctly + self.assertEqual( + usage_source.redshift_instance_type, RedshiftInstanceType.PROVISIONED + ) + + # Verify correct SQL statement selected + self.assertEqual( + usage_source.sql_stmt, + REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.PROVISIONED], + ) + + # CRITICAL: Verify filters use 'querytxt' (Provisioned column name) + self.assertIn("querytxt", usage_source.filters) + self.assertNotIn("query_text", usage_source.filters) + + # Verify specific filter patterns + self.assertIn("NOT ILIKE 'fetch%%'", usage_source.filters) + self.assertIn("NOT ILIKE 'padb_fetch_sample:%%'", usage_source.filters) + + @patch( + "metadata.ingestion.source.database.redshift.lineage.get_redshift_instance_type" + ) + def test_lineage_source_provisioned_initialization(self, mock_get_instance_type): + """Test RedshiftLineageSource filters and SQL statement for Provisioned""" + from metadata.ingestion.source.database.redshift.lineage import ( + RedshiftLineageSource, + ) + + mock_get_instance_type.return_value = RedshiftInstanceType.PROVISIONED + mock_engine = Mock() + + # Create instance without full initialization + lineage_source = RedshiftLineageSource.__new__(RedshiftLineageSource) + lineage_source.engine = mock_engine + lineage_source.redshift_instance_type = mock_get_instance_type.return_value + + # Simulate __init__ logic for filter and statement selection + if lineage_source.redshift_instance_type == RedshiftInstanceType.PROVISIONED: + lineage_source.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[ + RedshiftInstanceType.PROVISIONED + ] + lineage_source.filters = RedshiftLineageSource.provisioned_filters + + # Verify instance type detected correctly + self.assertEqual( + lineage_source.redshift_instance_type, RedshiftInstanceType.PROVISIONED + ) + + # Verify correct SQL statement selected + self.assertEqual( + lineage_source.sql_stmt, + REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.PROVISIONED], + ) + + # CRITICAL: Verify filters use 'querytxt' (Provisioned column name) + self.assertIn("querytxt", lineage_source.filters) + self.assertNotIn("query_text", lineage_source.filters) + + # Verify lineage-specific filter patterns + self.assertIn("ILIKE '%%create%%table%%as%%select%%'", lineage_source.filters) + self.assertIn("ILIKE '%%insert%%into%%select%%'", lineage_source.filters) + self.assertIn("ILIKE '%%update%%'", lineage_source.filters) + self.assertIn("ILIKE '%%merge%%'", lineage_source.filters) + + +if __name__ == "__main__": + unittest.main() diff --git a/ingestion/tests/unit/topology/database/test_redshift_serverless.py b/ingestion/tests/unit/topology/database/test_redshift_serverless.py index 025453870a19..16c373197ac6 100644 --- a/ingestion/tests/unit/topology/database/test_redshift_serverless.py +++ b/ingestion/tests/unit/topology/database/test_redshift_serverless.py @@ -21,14 +21,38 @@ from sqlalchemy.exc import ProgrammingError from metadata.ingestion.source.database.redshift.connection import ( - detect_redshift_serverless, + get_redshift_instance_type, ) +from metadata.ingestion.source.database.redshift.models import RedshiftInstanceType from metadata.ingestion.source.database.redshift.queries import ( - REDSHIFT_SERVERLESS_SQL_STATEMENT, - REDSHIFT_SQL_STATEMENT, - get_redshift_queries, + REDSHIFT_SQL_STATEMENT_MAP, ) -from metadata.ingestion.source.database.redshift.usage import RedshiftUsageSource + +# Mock Redshift configuration for testing +mock_redshift_config = { + "source": { + "type": "redshift", + "serviceName": "local_redshift_serverless", + "serviceConnection": { + "config": { + "type": "Redshift", + "username": "username", + "password": "password", + "database": "database", + "hostPort": "workgroup.account.region.redshift-serverless.amazonaws.com:5439", + } + }, + "sourceConfig": {"config": {"type": "DatabaseMetadata"}}, + }, + "sink": {"type": "metadata-rest", "config": {}}, + "workflowConfig": { + "openMetadataServerConfig": { + "hostPort": "http://localhost:8585/api", + "authProvider": "openmetadata", + "securityConfig": {"jwtToken": "redshift"}, + } + }, +} class TestRedshiftServerlessDetection(unittest.TestCase): @@ -42,134 +66,92 @@ def setUp(self): self.mock_connection ) - def test_detect_redshift_provisioned(self): - """Test detection of Redshift Provisioned cluster (STL tables accessible)""" - # Mock successful STL query execution - self.mock_connection.execute.return_value = None + def test_detect_serverless_when_stl_not_accessible(self): + """Test detection of Redshift Serverless when STL tables are not accessible (InsufficientPrivilege error)""" + # Mock InsufficientPrivilege error for STL query + self.mock_connection.execute.side_effect = ProgrammingError( + "permission denied for relation stl_query", None, InsufficientPrivilege() + ) - result = detect_redshift_serverless(self.mock_engine) + result = get_redshift_instance_type(self.mock_engine) - self.assertFalse(result) + self.assertEqual(result, RedshiftInstanceType.SERVERLESS) self.mock_connection.execute.assert_called_once() - def test_detect_redshift_serverless_insufficient_privilege(self): - """Test detection of Redshift Serverless (InsufficientPrivilege error)""" - # Mock InsufficientPrivilege error for STL query + def test_detect_serverless_generic_error(self): + """Test detection of Redshift Serverless on generic STL access error""" + # Mock generic error for STL query self.mock_connection.execute.side_effect = ProgrammingError( - "permission denied for relation stl_query", None, InsufficientPrivilege() + 'relation "stl_query" does not exist', {}, None ) - result = detect_redshift_serverless(self.mock_engine) + result = get_redshift_instance_type(self.mock_engine) - self.assertTrue(result) + self.assertEqual(result, RedshiftInstanceType.SERVERLESS) self.mock_connection.execute.assert_called_once() - def test_detect_redshift_serverless_generic_error(self): - """Test detection of Redshift Serverless (generic error)""" - # Mock generic error for STL query - self.mock_connection.execute.side_effect = Exception("Table does not exist") + def test_detect_provisioned_when_stl_accessible(self): + """Test detection of Redshift Provisioned cluster when STL tables are accessible""" + # Mock successful STL query execution + self.mock_connection.execute.return_value = None - result = detect_redshift_serverless(self.mock_engine) + result = get_redshift_instance_type(self.mock_engine) - self.assertTrue(result) + self.assertEqual(result, RedshiftInstanceType.PROVISIONED) self.mock_connection.execute.assert_called_once() - def test_get_redshift_queries_provisioned(self): - """Test query selection for Redshift Provisioned""" - queries = get_redshift_queries(is_serverless=False) + def test_serverless_uses_sys_queries(self): + """Test that Serverless uses SYS-based queries""" + sql = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.SERVERLESS] - self.assertEqual(queries["sql_statement"], REDSHIFT_SQL_STATEMENT) - self.assertIn("stl_query", queries["sql_statement"]) - self.assertIn("stl_querytext", queries["sql_statement"]) - self.assertIn("stl_scan", queries["sql_statement"]) + # Must use SYS views + self.assertIn("SYS_QUERY_HISTORY", sql) + self.assertIn("SYS_QUERY_TEXT", sql) + self.assertIn("SYS_QUERY_DETAIL", sql) - def test_get_redshift_queries_serverless(self): - """Test query selection for Redshift Serverless""" - queries = get_redshift_queries(is_serverless=True) + # Should NOT use STL tables + self.assertNotIn("stl_query", sql.lower()) + self.assertNotIn("stl_querytext", sql.lower()) + self.assertNotIn("stl_scan", sql.lower()) - self.assertEqual(queries["sql_statement"], REDSHIFT_SERVERLESS_SQL_STATEMENT) - self.assertIn("SYS_QUERY_HISTORY", queries["sql_statement"]) - self.assertIn("SYS_QUERY_DETAIL", queries["sql_statement"]) - self.assertNotIn("stl_query", queries["sql_statement"]) - self.assertNotIn("stl_querytext", queries["sql_statement"]) + # Check for proper filtering + self.assertIn("LOWER(status) = 'success'", sql) + self.assertIn("user_id > 1", sql) - @patch( - "metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless" - ) - def test_usage_source_serverless_initialization(self, mock_detect): - """Test RedshiftUsageSource initialization with Serverless detection""" - # Mock serverless detection - mock_detect.return_value = True - - # Mock config objects - _ = MagicMock() # config - _ = MagicMock() # metadata_config - - with patch.object( - RedshiftUsageSource, "__init__", return_value=None - ) as _mock_init: - # Create instance and manually set up attributes - usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) - usage_source.engine = self.mock_engine - usage_source.is_serverless = True - usage_source.sql_stmt = REDSHIFT_SERVERLESS_SQL_STATEMENT - usage_source.filters = ( - usage_source.serverless_filters - ) = """ - AND query_text NOT ILIKE 'fetch%%' - AND query_text NOT ILIKE 'padb_fetch_sample:%%' - AND query_text NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' - """ - - self.assertTrue(usage_source.is_serverless) - self.assertEqual(usage_source.sql_stmt, REDSHIFT_SERVERLESS_SQL_STATEMENT) - self.assertIn("query_text", usage_source.filters) + # Check for placeholder substitution + self.assertIn("{start_time}", sql) + self.assertIn("{end_time}", sql) + self.assertIn("{result_limit}", sql) + self.assertIn("{filters}", sql) - @patch( - "metadata.ingestion.source.database.redshift.usage.detect_redshift_serverless" - ) - def test_usage_source_provisioned_initialization(self, mock_detect): - """Test RedshiftUsageSource initialization with Provisioned detection""" - # Mock provisioned detection - mock_detect.return_value = False - - # Mock config objects - _ = MagicMock() # config - _ = MagicMock() # metadata_config - - with patch.object( - RedshiftUsageSource, "__init__", return_value=None - ) as _mock_init: - # Create instance and manually set up attributes - usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) - usage_source.engine = self.mock_engine - usage_source.is_serverless = False - usage_source.sql_stmt = REDSHIFT_SQL_STATEMENT - usage_source.filters = """ - AND querytxt NOT ILIKE 'fetch%%' - AND querytxt NOT ILIKE 'padb_fetch_sample:%%' - AND querytxt NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%' - """ - - self.assertFalse(usage_source.is_serverless) - self.assertEqual(usage_source.sql_stmt, REDSHIFT_SQL_STATEMENT) - self.assertIn("querytxt", usage_source.filters) + def test_provisioned_uses_stl_queries(self): + """Test that Provisioned uses STL-based queries""" + sql = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.PROVISIONED] + + # Must use STL tables + self.assertIn("stl_query", sql.lower()) + self.assertIn("stl_querytext", sql.lower()) + + # Should NOT use SYS views + self.assertNotIn("SYS_QUERY_HISTORY", sql) + self.assertNotIn("SYS_QUERY_TEXT", sql) + self.assertNotIn("SYS_QUERY_DETAIL", sql) def test_serverless_sql_statement_structure(self): """Test that the serverless SQL statement has the correct structure""" - statement = REDSHIFT_SERVERLESS_SQL_STATEMENT + statement = REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.SERVERLESS] # Check for SYS views self.assertIn("SYS_QUERY_HISTORY", statement) self.assertIn("SYS_QUERY_DETAIL", statement) # Check that STL views are not present - self.assertNotIn("stl_query", statement) - self.assertNotIn("stl_querytext", statement) - self.assertNotIn("stl_scan", statement) + self.assertNotIn("stl_query", statement.lower()) + self.assertNotIn("stl_querytext", statement.lower()) + self.assertNotIn("stl_scan", statement.lower()) # Check for proper filtering - self.assertIn("status = 'success'", statement) + self.assertIn("LOWER(status) = 'success'", statement) self.assertIn("user_id > 1", statement) # Check for placeholder substitution @@ -178,33 +160,100 @@ def test_serverless_sql_statement_structure(self): self.assertIn("{result_limit}", statement) self.assertIn("{filters}", statement) - def test_query_factory_returns_correct_types(self): - """Test that the query factory returns the expected dictionary structure""" - # Test provisioned queries - provisioned_queries = get_redshift_queries(is_serverless=False) - expected_keys = [ - "sql_statement", - "test_queries", - "table_changes", - "metrics_query", - ] - - for key in expected_keys: - self.assertIn(key, provisioned_queries) - self.assertIsInstance(provisioned_queries[key], str) - - # Test serverless queries - serverless_queries = get_redshift_queries(is_serverless=True) - - for key in expected_keys: - self.assertIn(key, serverless_queries) - self.assertIsInstance(serverless_queries[key], str) - - # Ensure they're different - self.assertNotEqual( - provisioned_queries["sql_statement"], serverless_queries["sql_statement"] + @patch( + "metadata.ingestion.source.database.redshift.usage.get_redshift_instance_type" + ) + def test_usage_source_serverless_filter_validation(self, mock_get_instance_type): + """Test that Serverless usage source uses correct filters with 'query_text' column""" + from metadata.ingestion.source.database.redshift.usage import ( + RedshiftUsageSource, ) + mock_get_instance_type.return_value = RedshiftInstanceType.SERVERLESS + + # Create instance without full initialization + usage_source = RedshiftUsageSource.__new__(RedshiftUsageSource) + usage_source.engine = self.mock_engine + usage_source.redshift_instance_type = mock_get_instance_type.return_value + + # Simulate __init__ logic for filter and statement selection + if usage_source.redshift_instance_type == RedshiftInstanceType.SERVERLESS: + usage_source.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[ + RedshiftInstanceType.SERVERLESS + ] + usage_source.filters = RedshiftUsageSource.serverless_filters + + # Verify instance type + self.assertEqual( + usage_source.redshift_instance_type, RedshiftInstanceType.SERVERLESS + ) + + # Verify SQL statement + self.assertEqual( + usage_source.sql_stmt, + REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.SERVERLESS], + ) + + # CRITICAL: Verify filters use 'query_text' (not 'querytxt') + self.assertIn("query_text", usage_source.filters) + self.assertNotIn("querytxt", usage_source.filters) + + # Verify specific filter patterns + self.assertIn("NOT ILIKE 'fetch%%'", usage_source.filters) + self.assertIn("NOT ILIKE 'padb_fetch_sample:%%'", usage_source.filters) + self.assertIn( + "NOT ILIKE 'Undoing%%transactions%%on%%table%%with%%current%%xid%%'", + usage_source.filters, + ) + + @patch( + "metadata.ingestion.source.database.redshift.lineage.get_redshift_instance_type" + ) + def test_lineage_source_serverless_filter_validation(self, mock_get_instance_type): + """Test that Serverless lineage source uses correct filters with 'query_text' column""" + from metadata.ingestion.source.database.redshift.lineage import ( + RedshiftLineageSource, + ) + + mock_get_instance_type.return_value = RedshiftInstanceType.SERVERLESS + + # Create instance without full initialization + lineage_source = RedshiftLineageSource.__new__(RedshiftLineageSource) + lineage_source.engine = self.mock_engine + lineage_source.redshift_instance_type = mock_get_instance_type.return_value + + # Simulate __init__ logic for filter and statement selection + if lineage_source.redshift_instance_type == RedshiftInstanceType.SERVERLESS: + lineage_source.sql_stmt = REDSHIFT_SQL_STATEMENT_MAP[ + RedshiftInstanceType.SERVERLESS + ] + lineage_source.filters = RedshiftLineageSource.serverless_filters + + # Verify instance type + self.assertEqual( + lineage_source.redshift_instance_type, RedshiftInstanceType.SERVERLESS + ) + + # Verify SQL statement + self.assertEqual( + lineage_source.sql_stmt, + REDSHIFT_SQL_STATEMENT_MAP[RedshiftInstanceType.SERVERLESS], + ) + + # CRITICAL: Verify filters use 'query_text' (not 'querytxt') + self.assertIn("query_text", lineage_source.filters) + self.assertNotIn("querytxt", lineage_source.filters) + + # Verify lineage-specific patterns with query_type column + self.assertIn("ILIKE '%%create%%table%%as%%select%%'", lineage_source.filters) + self.assertIn("query_type = 'CTAS'", lineage_source.filters) + self.assertIn("ILIKE '%%insert%%into%%select%%'", lineage_source.filters) + self.assertIn("query_type = 'INSERT'", lineage_source.filters) + self.assertIn("ILIKE '%%update%%'", lineage_source.filters) + self.assertIn("query_type = 'UPDATE'", lineage_source.filters) + self.assertIn("ILIKE '%%merge%%'", lineage_source.filters) + self.assertIn("query_type = 'MERGE'", lineage_source.filters) + if __name__ == "__main__": unittest.main() diff --git a/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md b/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md index fb8452206762..c973a1b54327 100644 --- a/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md +++ b/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md @@ -18,9 +18,14 @@ GRANT SELECT ON TABLE svv_table_info to test_user; Executing the profiler Workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. The user should also be allowed to view information in `svv_table_info` for all objects in the database. More information on the profiler workflow setup can be found here and data quality tests here. +Information on **System Metrics** profiling can be found here. + ### Usage & Lineage -For the usage and lineage workflow, the user will need `SELECT` privilege on `STL_QUERY` table. You can find more information on the usage workflow here and the lineage workflow here. +For the usage and lineage workflow, the user will need `SELECT` privilege on: +- `STL_QUERY`, `STL_QUERYTEXT`, `STL_SCAN` and `SVL_STORED_PROC_CALL` views for Provisioned cluster +- `SYS_QUERY_HISTORY`, `SYS_QUERY_TEXT`, `SYS_QUERY_DETAIL` and `SYS_PROCEDURE_CALL` for Serverless instance. +You can find more information on the usage workflow here and the lineage workflow here. You can find further information on the Redshift connector in the docs. From bfc398dd3eba671e8076f71507b48ba95042807a Mon Sep 17 00:00:00 2001 From: Mohit Tilala Date: Mon, 22 Dec 2025 21:00:58 +0530 Subject: [PATCH 4/5] Refine test get queries flow --- .../source/database/redshift/connection.py | 27 +++++++------------ 1 file changed, 9 insertions(+), 18 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py index 828683cd69e9..26eb2f382839 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/connection.py @@ -124,24 +124,15 @@ def test_get_queries_permissions(engine_: Engine): redshift_instance_type = get_redshift_instance_type(engine_) with engine_.connect() as conn: - if redshift_instance_type == RedshiftInstanceType.PROVISIONED: - res = conn.execute( - REDSHIFT_TEST_GET_QUERIES_MAP[RedshiftInstanceType.PROVISIONED] - ).fetchone() - if not all(res): - raise SourceConnectionException( - "We don't have the right permissions to list queries from stl views (Redshift Provisioned)" - f" - {res}" - ) - else: - res = conn.execute( - REDSHIFT_TEST_GET_QUERIES_MAP[RedshiftInstanceType.SERVERLESS] - ).fetchone() - if not all(res): - raise SourceConnectionException( - "We don't have the right permissions to list queries from sys views (Redshift Serverless)" - f" - {res}" - ) + res = conn.execute( + REDSHIFT_TEST_GET_QUERIES_MAP[redshift_instance_type] + ).fetchone() + if not all(res): + raise SourceConnectionException( + f"We don't have the right permissions to list queries from sys views (Redshift Serverless) - {res}" + if redshift_instance_type == RedshiftInstanceType.SERVERLESS + else f"We don't have the right permissions to list queries from stl views (Redshift Provisioned) - {res}" # noqa: E501 + ) test_fn = { "CheckAccess": partial(test_connection_engine_step, engine), From db8b2cd85531f90045bb84dd1d9a05c4dc8afd09 Mon Sep 17 00:00:00 2001 From: Mohit Tilala Date: Mon, 22 Dec 2025 21:12:00 +0530 Subject: [PATCH 5/5] Update docs and test for lineage/usage queries statement --- .../metadata/ingestion/source/database/redshift/queries.py | 1 - .../json/data/testConnections/database/redshift.json | 4 ++-- .../resources/ui/public/locales/en-US/Database/Redshift.md | 4 ++-- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py index 17f555b6c851..4ba4525f9cab 100644 --- a/ingestion/src/metadata/ingestion/source/database/redshift/queries.py +++ b/ingestion/src/metadata/ingestion/source/database/redshift/queries.py @@ -309,7 +309,6 @@ REDSHIFT_SERVERLESS_TEST_GET_QUERIES = """ SELECT - has_table_privilege('SVV_TABLE_INFO', 'SELECT') as can_access_svv_table_info, has_table_privilege('SYS_QUERY_HISTORY', 'SELECT') as can_access_sys_query_history, has_table_privilege('SYS_QUERY_TEXT', 'SELECT') as can_access_sys_query_text, has_table_privilege('SYS_QUERY_DETAIL', 'SELECT') as can_access_sys_query_detail, diff --git a/openmetadata-service/src/main/resources/json/data/testConnections/database/redshift.json b/openmetadata-service/src/main/resources/json/data/testConnections/database/redshift.json index b81fec0f3f27..a2660393232f 100644 --- a/openmetadata-service/src/main/resources/json/data/testConnections/database/redshift.json +++ b/openmetadata-service/src/main/resources/json/data/testConnections/database/redshift.json @@ -42,8 +42,8 @@ }, { "name": "GetQueries", - "description": "Check if we can access the pg_catalog.svv_table_info & pg_catalog.stl_query tables to get query logs, These queries are analyzed in the usage & lineage workflow.", - "errorMessage": "Failed to fetch queries, please validate if user has select privilege for tables pg_catalog.svv_table_info & pg_catalog.stl_query to get query logs.", + "description": "Check if we can access the `SVV_TABLE_INFO`, `STL_QUERY`, `STL_QUERYTEXT`, `STL_SCAN` and `SVL_STORED_PROC_CALL` (Provisioned Cluster) OR `SYS_QUERY_HISTORY`, `SYS_QUERY_TEXT`, `SYS_QUERY_DETAIL` and `SYS_PROCEDURE_CALL` (Serverless Cluster) to get query logs, These queries are analyzed in the usage & lineage workflow.", + "errorMessage": "Failed to fetch queries, please validate if user has select privilege for tables `SVV_TABLE_INFO`, `STL_QUERY`, `STL_QUERYTEXT`, `STL_SCAN` and `SVL_STORED_PROC_CALL` (Provisioned Cluster) OR `SYS_QUERY_HISTORY`, `SYS_QUERY_TEXT`, `SYS_QUERY_DETAIL` and `SYS_PROCEDURE_CALL` (Serverless Cluster) to get query logs.", "mandatory": false } ] diff --git a/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md b/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md index c973a1b54327..e7b42242cbac 100644 --- a/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md +++ b/openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Redshift.md @@ -16,14 +16,14 @@ GRANT SELECT ON TABLE svv_table_info to test_user; ### Profiler & Data Quality -Executing the profiler Workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. The user should also be allowed to view information in `svv_table_info` for all objects in the database. More information on the profiler workflow setup can be found here and data quality tests here. +Executing the profiler Workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. The user should also be allowed to view information in `SVV_TABLE_INFO` for all objects in the database. More information on the profiler workflow setup can be found here and data quality tests here. Information on **System Metrics** profiling can be found here. ### Usage & Lineage For the usage and lineage workflow, the user will need `SELECT` privilege on: -- `STL_QUERY`, `STL_QUERYTEXT`, `STL_SCAN` and `SVL_STORED_PROC_CALL` views for Provisioned cluster +- `SVV_TABLE_INFO`, `STL_QUERY`, `STL_QUERYTEXT`, `STL_SCAN` and `SVL_STORED_PROC_CALL` views for Provisioned cluster - `SYS_QUERY_HISTORY`, `SYS_QUERY_TEXT`, `SYS_QUERY_DETAIL` and `SYS_PROCEDURE_CALL` for Serverless instance. You can find more information on the usage workflow here and the lineage workflow here.