Fix: [Redshift] interval and numeric column type parsing (#27983)

* Fix: Redshift interval and numeric column type parsing

* py_format

* Fix numeric args fallback to _init_args when charlen is None

* Add regression test for numeric(P, S) with space falling back to _init_args

Author: mohittilala

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

@@ -234,9 +234,11 @@ def _get_kwargs_for_time_type(kwargs, charlen, attype):
 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 = tuple(int(p) for p in args)
 
     elif attype == "double precision":
         args = (53,)
@@ -257,6 +259,7 @@ def _get_args_and_kwargs(charlen, attype, format_type):
             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

@@ -368,5 +368,110 @@ def test_numeric_and_character_varying_positional_arguments_are_unchanged(self):
         self.assertEqual(varchar_kwargs, {})
 
 
+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, {})
+
+    def test_numeric_with_space_after_comma_falls_back_to_init_args(self):
+        """numeric(P, S) with a space must still parse precision and scale."""
+        args, kwargs = _get_args_and_kwargs(None, "numeric", "numeric(10, 2)")
+
+        self.assertEqual(args, (10, 2))
+        self.assertEqual(kwargs, {})
+
+
 if __name__ == "__main__":
     unittest.main()