fix(greenplum): default to v6 on version detection failure; add table-listing query branch tests

Address reviewer feedback on PR open-metadata#27288:

- Default to Greenplum 6 (conservative) when version detection fails or output is unparseable, since GP7 queries reference catalogs/columns that do not exist on GP6 (was defaulting to 7, which could regress GP6 clusters).

- Add unit tests verifying query_table_names_and_types selects GREENPLUM_GET_TABLE_NAMES_V6 vs GREENPLUM_GET_TABLE_NAMES_V7 based on detected major version, covering the table discovery path that originally failed in open-metadata#24559.

- Apply ruff format to satisfy py-checkstyle CI.

Author: RajdeepKushwaha5

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

@@ -123,40 +123,32 @@ def _get_greenplum_major_version(self) -> int:
                 match = re.search(r"Greenplum Database (\d+)", version_string)
                 if match:
                     self._greenplum_version = int(match.group(1))
-                    logger.info(
-                        "Detected Greenplum major version %d", self._greenplum_version
-                    )
+                    logger.info("Detected Greenplum major version %d", self._greenplum_version)
                 else:
                     logger.warning(
-                        "Could not parse Greenplum major version from"
-                        " SELECT version() output, defaulting to 7"
+                        "Could not parse Greenplum major version from SELECT version() output, "
+                        "defaulting conservatively to 6"
                     )
                     logger.debug("Full version() output: %s", version_string)
-                    self._greenplum_version = 7
+                    self._greenplum_version = 6
             except SQLAlchemyError as exc:
                 logger.debug(traceback.format_exc())
                 logger.warning(
-                    "Could not determine Greenplum version (%s), defaulting to 7",
+                    "Could not determine Greenplum version (%s), defaulting conservatively to 6",
                     exc,
                 )
-                self._greenplum_version = 7
+                self._greenplum_version = 6
         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]:
+    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
-        )
+        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(table_names_query),
             {"schema": schema_name},

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

@@ -63,9 +63,7 @@
     },
 }
 
-PartitionRow = namedtuple(
-    "PartitionRow", ["schema", "table_name", "partition_strategy", "column_name"]
-)
+PartitionRow = namedtuple("PartitionRow", ["schema", "table_name", "partition_strategy", "column_name"])
 
 
 class greenplumUnitTest(TestCase):
@@ -91,8 +89,7 @@ def test_version_detection_v6(self):
         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.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"
+            "PostgreSQL 9.4.26 (Greenplum Database 6.25.3 build dev) on x86_64-pc-linux-gnu compiled by gcc 6.4.0"
         )
         self.greenplum_source.engine = mock_engine
 
@@ -106,8 +103,7 @@ def test_version_detection_v7(self):
         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.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"
+            "PostgreSQL 12.12 (Greenplum Database 7.0.0 build dev) on x86_64-pc-linux-gnu compiled by gcc 12.3.1"
         )
         self.greenplum_source.engine = mock_engine
 
@@ -121,8 +117,7 @@ def test_version_detection_caches_result(self):
         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.scalar.return_value = (
-            "PostgreSQL 12.12 (Greenplum Database 7.2.0 build dev) on"
-            " x86_64-pc-linux-gnu"
+            "PostgreSQL 12.12 (Greenplum Database 7.2.0 build dev) on x86_64-pc-linux-gnu"
         )
         self.greenplum_source.engine = mock_engine
 
@@ -131,28 +126,24 @@ def test_version_detection_caches_result(self):
         self.greenplum_source._get_greenplum_major_version()
         mock_engine.connect.assert_called_once()
 
-    def test_version_detection_defaults_to_7_on_error(self):
+    def test_version_detection_defaults_to_6_on_error(self):
         mock_engine = MagicMock()
-        mock_engine.connect.side_effect = OperationalError(
-            "SELECT version()", {}, Exception("connection error")
-        )
+        mock_engine.connect.side_effect = OperationalError("SELECT version()", {}, Exception("connection error"))
         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._get_greenplum_major_version() == 6
 
-    def test_version_detection_defaults_to_7_on_unparseable(self):
+    def test_version_detection_defaults_to_6_on_unparseable(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.scalar.return_value = (
-            "PostgreSQL 14.0 on x86_64-pc-linux-gnu"
-        )
+        mock_conn.execute.return_value.scalar.return_value = "PostgreSQL 14.0 on x86_64-pc-linux-gnu"
         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._get_greenplum_major_version() == 6
 
     def test_partition_details_v7(self):
         mock_engine = MagicMock()
@@ -172,9 +163,7 @@ def test_partition_details_v7(self):
         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()
-        )
+        is_partitioned, partition = self.greenplum_source.get_table_partition_details("sales", "public", MagicMock())
 
         assert is_partitioned is True
         assert partition == TablePartition(
@@ -205,9 +194,7 @@ def test_partition_details_v6(self):
         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()
-        )
+        is_partitioned, partition = self.greenplum_source.get_table_partition_details("sales", "public", MagicMock())
 
         assert is_partitioned is True
         assert partition == TablePartition(
@@ -258,9 +245,50 @@ def test_partition_details_expression_key_returns_true_no_columns(self):
         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()
-        )
+        is_partitioned, partition = self.greenplum_source.get_table_partition_details("sales", "public", MagicMock())
 
         assert is_partitioned is True
         assert partition is None
+
+    def test_query_table_names_uses_v6_query_for_v6(self):
+        """Issue #24559: ensure GP6 uses pg_partition_rule-based table-listing query."""
+        from metadata.ingestion.source.database.greenplum.queries import (
+            GREENPLUM_GET_TABLE_NAMES_V6,
+        )
+
+        mock_connection = MagicMock()
+        mock_connection.execute.return_value = []
+        self.greenplum_source._greenplum_version = 6
+
+        with patch.object(
+            type(self.greenplum_source),
+            "connection",
+            new_callable=lambda: property(lambda self: mock_connection),
+        ):
+            list(self.greenplum_source.query_table_names_and_types("public"))
+
+        executed_sql = str(mock_connection.execute.call_args[0][0])
+        assert executed_sql.strip() == GREENPLUM_GET_TABLE_NAMES_V6.strip()
+        assert "pg_partition_rule" in executed_sql
+
+    def test_query_table_names_uses_v7_query_for_v7(self):
+        """Issue #24559: ensure GP7 uses relispartition-based table-listing query."""
+        from metadata.ingestion.source.database.greenplum.queries import (
+            GREENPLUM_GET_TABLE_NAMES_V7,
+        )
+
+        mock_connection = MagicMock()
+        mock_connection.execute.return_value = []
+        self.greenplum_source._greenplum_version = 7
+
+        with patch.object(
+            type(self.greenplum_source),
+            "connection",
+            new_callable=lambda: property(lambda self: mock_connection),
+        ):
+            list(self.greenplum_source.query_table_names_and_types("public"))
+
+        executed_sql = str(mock_connection.execute.call_args[0][0])
+        assert executed_sql.strip() == GREENPLUM_GET_TABLE_NAMES_V7.strip()
+        assert "relispartition" in executed_sql
+        assert "pg_partition_rule" not in executed_sql