Fixes #24559: Add Greenplum 7 partition support

Greenplum 7 adopted PostgreSQL 10+ declarative partitioning, replacing the
GP-specific pg_partition and pg_partition_rule catalog tables with the
standard pg_partitioned_table and relispartition column in pg_class.

The connector now detects the Greenplum major version at runtime by parsing
SELECT version() output with a regex, and selects the appropriate queries:

- GP6: pg_partition_rule JOIN for table names, pg_partition with
  parkind/paratts for partition details (existing behavior)
- GP7+: relispartition=false for table names, pg_partitioned_table with
  partstrat/partattrs for partition details (PostgreSQL-compatible)

The version is cached per source instance to avoid repeated queries.

Author: RajdeepKushwaha5

ingestion/src/metadata/ingestion/source/database/greenplum/metadata.py

@@ -11,6 +11,8 @@
 """
 Greenplum source module
 """
+
+import re
 import traceback
 from collections import namedtuple
 from typing import Iterable, Optional, Tuple
@@ -47,8 +49,11 @@
 )
 from metadata.ingestion.source.database.greenplum.queries import (
     GREENPLUM_GET_DB_NAMES,
-    GREENPLUM_GET_TABLE_NAMES,
-    GREENPLUM_PARTITION_DETAILS,
+    GREENPLUM_GET_TABLE_NAMES_V6,
+    GREENPLUM_GET_TABLE_NAMES_V7,
+    GREENPLUM_GET_VERSION,
+    GREENPLUM_PARTITION_DETAILS_V6,
+    GREENPLUM_PARTITION_DETAILS_V7,
 )
 from metadata.ingestion.source.database.greenplum.utils import (
     get_column_info,
@@ -91,6 +96,10 @@ class GreenplumSource(CommonDbSourceService, MultiDBSource):
     Database metadata from Greenplum Source
     """
 
+    def __init__(self, config, metadata):
+        super().__init__(config, metadata)
+        self._greenplum_version: Optional[int] = None
+
     @classmethod
     def create(
         cls,
@@ -106,15 +115,45 @@ def create(
             )
         return cls(config, metadata)
 
+    def _get_greenplum_major_version(self) -> int:
+        if self._greenplum_version is None:
+            try:
+                with self.engine.connect() as conn:
+                    result = conn.execute(text(GREENPLUM_GET_VERSION))
+                    version_string = result.scalar()
+                match = re.search(r"Greenplum Database (\d+)", version_string)
+                if match:
+                    self._greenplum_version = int(match.group(1))
+                else:
+                    logger.warning(
+                        "Could not parse Greenplum version from '%s',"
+                        " defaulting to 7",
+                        version_string,
+                    )
+                    self._greenplum_version = 7
+            except Exception:
+                logger.debug(traceback.format_exc())
+                logger.warning("Could not determine Greenplum version, defaulting to 7")
+                self._greenplum_version = 7
+        return self._greenplum_version
+
+    def _is_v7_or_later(self) -> bool:
+        return self._get_greenplum_major_version() >= 7
+
     def query_table_names_and_types(
         self, schema_name: str
     ) -> Iterable[TableNameAndType]:
         """
         Overwrite the inspector implementation to handle partitioned
         and foreign types
         """
+        table_names_query = (
+            GREENPLUM_GET_TABLE_NAMES_V7
+            if self._is_v7_or_later()
+            else GREENPLUM_GET_TABLE_NAMES_V6
+        )
         result = self.connection.execute(
-            sql.text(GREENPLUM_GET_TABLE_NAMES),
+            sql.text(table_names_query),
             {"schema": schema_name},
         )
 
@@ -149,9 +188,11 @@ def get_database_names(self) -> Iterable[str]:
 
                 if filter_by_database(
                     self.source_config.databaseFilterPattern,
-                    database_fqn
-                    if self.source_config.useFqnForFiltering
-                    else new_database,
+                    (
+                        database_fqn
+                        if self.source_config.useFqnForFiltering
+                        else new_database
+                    ),
                 ):
                     self.status.filter(database_fqn, "Database Filtered Out")
                     continue
@@ -167,11 +208,18 @@ def get_database_names(self) -> Iterable[str]:
 
     def get_table_partition_details(
         self, table_name: str, schema_name: str, inspector: Inspector
+    ) -> Tuple[bool, Optional[TablePartition]]:
+        if self._is_v7_or_later():
+            return self._get_table_partition_details_v7(table_name, schema_name)
+        return self._get_table_partition_details_v6(table_name, schema_name)
+
+    def _get_table_partition_details_v6(
+        self, table_name: str, schema_name: str
     ) -> Tuple[bool, Optional[TablePartition]]:
         with self.engine.connect() as conn:
             result = conn.execute(
                 text(
-                    GREENPLUM_PARTITION_DETAILS.format(
+                    GREENPLUM_PARTITION_DETAILS_V6.format(
                         table_name=table_name, schema_name=schema_name
                     )
                 )
@@ -194,3 +242,30 @@ def get_table_partition_details(
             )
             return True, partition_details
         return False, None
+
+    def _get_table_partition_details_v7(
+        self, table_name: str, schema_name: str
+    ) -> Tuple[bool, Optional[TablePartition]]:
+        with self.engine.connect() as conn:
+            result = conn.execute(
+                text(GREENPLUM_PARTITION_DETAILS_V7),
+                {"table_name": table_name, "schema_name": schema_name},
+            ).all()
+
+        if result:
+            partition_details = TablePartition(
+                columns=[
+                    PartitionColumnDetails(
+                        columnName=row.column_name,
+                        intervalType=INTERVAL_TYPE_MAP.get(
+                            result[0].partition_strategy,
+                            PartitionIntervalTypes.COLUMN_VALUE,
+                        ),
+                        interval=None,
+                    )
+                    for row in result
+                    if row.column_name
+                ]
+            )
+            return True, partition_details
+        return False, None

ingestion/src/metadata/ingestion/source/database/greenplum/queries.py

@@ -16,7 +16,9 @@
 
 # https://www.postgresql.org/docs/current/catalog-pg-class.html
 # r = ordinary table, v = view, m = materialized view, c = composite type, f = foreign table, p = partitioned table,
-GREENPLUM_GET_TABLE_NAMES = """
+
+# Greenplum 6: uses pg_partition_rule to filter out child partitions
+GREENPLUM_GET_TABLE_NAMES_V6 = """
     select c.relname, c.relkind
     from pg_catalog.pg_class c
         left outer join pg_catalog.pg_partition_rule pr on c.oid = pr.parchildrelid
@@ -26,8 +28,15 @@
         and n.nspname = :schema
 """
 
-GREENPLUM_PARTITION_DETAILS = textwrap.dedent(
-    """
+# Greenplum 7+: uses PostgreSQL-style relispartition to filter out child partitions
+GREENPLUM_GET_TABLE_NAMES_V7 = """
+    SELECT c.relname, c.relkind FROM pg_class c
+    JOIN pg_namespace n ON n.oid = c.relnamespace
+    WHERE n.nspname = :schema AND c.relkind in ('r', 'p', 'f') AND relispartition = false
+"""
+
+# Greenplum 6: uses pg_partition catalog with parkind/paratts columns
+GREENPLUM_PARTITION_DETAILS_V6 = textwrap.dedent("""
     select
         ns.nspname as schema,
         par.relname as table_name,
@@ -57,8 +66,38 @@
         and col.table_name = par.relname
         and ordinal_position = pt.column_index
     where par.relname='{table_name}' and  ns.nspname='{schema_name}'
-    """
-)
+    """)
+
+# Greenplum 7+: uses PostgreSQL-standard pg_partitioned_table with partstrat/partattrs
+GREENPLUM_PARTITION_DETAILS_V7 = textwrap.dedent("""
+    select
+        par.relnamespace::regnamespace::text as schema,
+        par.relname as table_name,
+        partition_strategy,
+        col.column_name
+    from
+        (select
+             partrelid,
+             partnatts,
+             case partstrat
+                  when 'l' then 'list'
+                  when 'h' then 'hash'
+                  when 'r' then 'range' end as partition_strategy,
+             unnest(partattrs) column_index
+         from
+             pg_partitioned_table) pt
+    join
+        pg_class par
+    on
+        par.oid = pt.partrelid
+    left join
+        information_schema.columns col
+    on
+        col.table_schema = par.relnamespace::regnamespace::text
+        and col.table_name = par.relname
+        and ordinal_position = pt.column_index
+     where par.relname=:table_name and  par.relnamespace::regnamespace::text=:schema_name
+    """)
 
 GREENPLUM_TABLE_COMMENTS = """
     SELECT n.nspname as schema,
@@ -139,3 +178,7 @@
 GREENPLUM_GET_SERVER_VERSION = """
 show server_version
 """
+
+GREENPLUM_GET_VERSION = """
+SELECT version()
+"""