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,44 @@ def create(
             )
         return cls(config, metadata)
 
+    def _get_greenplum_major_version(self) -> int:
+        if self._greenplum_version is None:
+            try:
+                result = self.engine.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 +187,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 +207,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 +241,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()
+"""

ingestion/tests/unit/topology/database/test_greenplum.py

@@ -13,9 +13,15 @@
 Test Greenplum using the topology
 """
 
+from collections import namedtuple
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
+from metadata.generated.schema.entity.data.table import (
+    PartitionColumnDetails,
+    PartitionIntervalTypes,
+    TablePartition,
+)
 from metadata.generated.schema.metadataIngestion.workflow import (
     OpenMetadataWorkflowConfig,
 )
@@ -55,6 +61,10 @@
     },
 }
 
+PartitionRow = namedtuple(
+    "PartitionRow", ["schema", "table_name", "partition_strategy", "column_name"]
+)
+
 
 class greenplumUnitTest(TestCase):
     @patch(
@@ -76,3 +86,147 @@ def __init__(self, methodName, test_connection) -> None:
     def test_close_connection(self, engine, connection):
         connection.return_value = True
         self.greenplum_source.close()
+
+    def test_version_detection_v6(self):
+        mock_engine = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalar.return_value = (
+            "PostgreSQL 9.4.26 (Greenplum Database 6.25.3 build dev) on"
+            " x86_64-pc-linux-gnu compiled by gcc 6.4.0"
+        )
+        mock_engine.execute.return_value = mock_result
+        self.greenplum_source.engine = mock_engine
+
+        self.greenplum_source._greenplum_version = None
+        assert self.greenplum_source._get_greenplum_major_version() == 6
+        assert not self.greenplum_source._is_v7_or_later()
+
+    def test_version_detection_v7(self):
+        mock_engine = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalar.return_value = (
+            "PostgreSQL 12.12 (Greenplum Database 7.0.0 build dev) on"
+            " x86_64-pc-linux-gnu compiled by gcc 12.3.1"
+        )
+        mock_engine.execute.return_value = mock_result
+        self.greenplum_source.engine = mock_engine
+
+        self.greenplum_source._greenplum_version = None
+        assert self.greenplum_source._get_greenplum_major_version() == 7
+        assert self.greenplum_source._is_v7_or_later()
+
+    def test_version_detection_caches_result(self):
+        mock_engine = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalar.return_value = (
+            "PostgreSQL 12.12 (Greenplum Database 7.2.0 build dev) on"
+            " x86_64-pc-linux-gnu"
+        )
+        mock_engine.execute.return_value = mock_result
+        self.greenplum_source.engine = mock_engine
+
+        self.greenplum_source._greenplum_version = None
+        self.greenplum_source._get_greenplum_major_version()
+        self.greenplum_source._get_greenplum_major_version()
+        mock_engine.execute.assert_called_once()
+
+    def test_version_detection_defaults_to_7_on_error(self):
+        mock_engine = MagicMock()
+        mock_engine.execute.side_effect = Exception("connection error")
+        self.greenplum_source.engine = mock_engine
+
+        self.greenplum_source._greenplum_version = None
+        assert self.greenplum_source._get_greenplum_major_version() == 7
+
+    def test_version_detection_defaults_to_7_on_unparseable(self):
+        mock_engine = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalar.return_value = "PostgreSQL 14.0 on x86_64-pc-linux-gnu"
+        mock_engine.execute.return_value = mock_result
+        self.greenplum_source.engine = mock_engine
+
+        self.greenplum_source._greenplum_version = None
+        assert self.greenplum_source._get_greenplum_major_version() == 7
+
+    def test_partition_details_v7(self):
+        mock_engine = MagicMock()
+        mock_conn = MagicMock()
+        mock_engine.connect.return_value.__enter__ = MagicMock(return_value=mock_conn)
+        mock_engine.connect.return_value.__exit__ = MagicMock(return_value=False)
+
+        mock_conn.execute.return_value.all.return_value = [
+            PartitionRow(
+                schema="public",
+                table_name="sales",
+                partition_strategy="range",
+                column_name="sale_date",
+            )
+        ]
+
+        self.greenplum_source.engine = mock_engine
+        self.greenplum_source._greenplum_version = 7
+
+        is_partitioned, partition = self.greenplum_source.get_table_partition_details(
+            "sales", "public", MagicMock()
+        )
+
+        assert is_partitioned is True
+        assert partition == TablePartition(
+            columns=[
+                PartitionColumnDetails(
+                    columnName="sale_date",
+                    intervalType=PartitionIntervalTypes.TIME_UNIT,
+                    interval=None,
+                )
+            ]
+        )
+
+    def test_partition_details_v6(self):
+        mock_engine = MagicMock()
+        mock_conn = MagicMock()
+        mock_engine.connect.return_value.__enter__ = MagicMock(return_value=mock_conn)
+        mock_engine.connect.return_value.__exit__ = MagicMock(return_value=False)
+
+        mock_conn.execute.return_value.all.return_value = [
+            PartitionRow(
+                schema="public",
+                table_name="sales",
+                partition_strategy="list",
+                column_name="region",
+            )
+        ]
+
+        self.greenplum_source.engine = mock_engine
+        self.greenplum_source._greenplum_version = 6
+
+        is_partitioned, partition = self.greenplum_source.get_table_partition_details(
+            "sales", "public", MagicMock()
+        )
+
+        assert is_partitioned is True
+        assert partition == TablePartition(
+            columns=[
+                PartitionColumnDetails(
+                    columnName="region",
+                    intervalType=PartitionIntervalTypes.COLUMN_VALUE,
+                    interval=None,
+                )
+            ]
+        )
+
+    def test_partition_details_no_results(self):
+        mock_engine = MagicMock()
+        mock_conn = MagicMock()
+        mock_engine.connect.return_value.__enter__ = MagicMock(return_value=mock_conn)
+        mock_engine.connect.return_value.__exit__ = MagicMock(return_value=False)
+        mock_conn.execute.return_value.all.return_value = []
+
+        self.greenplum_source.engine = mock_engine
+        self.greenplum_source._greenplum_version = 7
+
+        is_partitioned, partition = self.greenplum_source.get_table_partition_details(
+            "not_partitioned", "public", MagicMock()
+        )
+
+        assert is_partitioned is False
+        assert partition is None