Fix: [Redshift] interval and numeric column type parsing

ingestion/src/metadata/ingestion/source/database/redshift/utils.py

@@ -87,8 +87,8 @@
 
 
 def get_multi_columns(
     self,
     connection,
     schema: str | None = None,
     filter_names: Any | None = None,
     scope: Any | None = None,
@@ -234,9 +234,11 @@
 def _get_args_and_kwargs(charlen, attype, format_type):
     kwargs = {}
     args = _init_args(format_type)
-    if attype == "numeric" and charlen:
-        prec, scale = charlen.split(",")
-        args = (int(prec), int(scale))
+    if attype == "numeric":
+        if charlen:
+            args = tuple(int(p) for p in charlen.split(","))
+        else:
+            args = ()
 
     elif attype == "double precision":
         args = (53,)
@@ -256,6 +258,7 @@
             args = (int(charlen),)
 
     elif attype.startswith("interval"):
+        args = ()
         field_match = re.match(r"interval (.+)", attype, re.I)
         if charlen:
             kwargs["precision"] = int(charlen)

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

@@ -16,7 +16,10 @@
 
 from metadata.ingestion.source.database.redshift.utils import (
     _get_all_relation_info,
+    _get_args_and_kwargs,
+    _update_coltype,
     get_view_definition,
+    ischema_names,
 )
 
 
@@ -294,5 +297,105 @@ def test_cache_invalidates_on_schema_change(self):
         self.assertEqual(self.mock_connection.execute.call_count, 2)
 
 
+class TestRedshiftIntervalParsing(unittest.TestCase):
+    """Test Redshift interval column type argument parsing."""
+
+    def test_interval_with_precision_uses_keyword_argument(self):
+        """interval(N) must route precision through kwargs, not positional args."""
+        args, kwargs = _get_args_and_kwargs("6", "interval", "interval(6)")
+
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {"precision": 6})
+
+        coltype = _update_coltype(
+            ischema_names["interval"],
+            args,
+            kwargs,
+            "interval",
+            "duration",
+            False,
+        )
+
+        self.assertEqual(coltype.precision, 6)
+        self.assertIsNone(coltype.fields)
+
+    def test_interval_with_fields_and_precision_uses_keyword_arguments(self):
+        """interval <fields>(N) must route both precision and fields through kwargs."""
+        args, kwargs = _get_args_and_kwargs(
+            "6", "interval day to second", "interval day to second(6)"
+        )
+
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {"precision": 6, "fields": "day to second"})
+
+        coltype = _update_coltype(
+            ischema_names["interval"],
+            args,
+            kwargs,
+            "interval",
+            "duration",
+            False,
+        )
+
+        self.assertEqual(coltype.precision, 6)
+        self.assertEqual(coltype.fields, "day to second")
+
+    def test_interval_without_precision_keeps_args_empty(self):
+        """Bare interval must produce empty args and empty kwargs."""
+        args, kwargs = _get_args_and_kwargs(None, "interval", "interval")
+
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {})
+
+
+class TestRedshiftNumericParsing(unittest.TestCase):
+    """Test Redshift numeric column type argument parsing."""
+
+    def test_numeric_with_precision_only_does_not_crash(self):
+        """numeric(N) without scale must parse precision-only without ValueError."""
+        args, kwargs = _get_args_and_kwargs("10", "numeric", "numeric(10)")
+
+        self.assertEqual(args, (10,))
+        self.assertEqual(kwargs, {})
+
+        coltype = _update_coltype(
+            ischema_names["numeric"],
+            args,
+            kwargs,
+            "numeric",
+            "amount",
+            False,
+        )
+
+        self.assertEqual(coltype.precision, 10)
+        self.assertIsNone(coltype.scale)
+
+    def test_numeric_with_precision_and_scale_unchanged(self):
+        """Regression: numeric(P,S) must continue to parse both as positional args."""
+        args, kwargs = _get_args_and_kwargs("10,2", "numeric", "numeric(10,2)")
+
+        self.assertEqual(args, (10, 2))
+        self.assertEqual(kwargs, {})
+
+        coltype = _update_coltype(
+            ischema_names["numeric"],
+            args,
+            kwargs,
+            "numeric",
+            "amount",
+            False,
+        )
+
+        self.assertEqual(coltype.precision, 10)
+        self.assertEqual(coltype.scale, 2)
+
+    def test_numeric_without_charlen_keeps_args_empty(self):
+        """Bare numeric must produce empty args and empty kwargs."""
+        args, kwargs = _get_args_and_kwargs(None, "numeric", "numeric")
+
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {})
+
+
 if __name__ == "__main__":
     unittest.main()