[FIX] util.column_type: optionally return the max length of varchar colums

Returning the max length in ff3001a was
a good idea, but that changed the API and break a few existing scripts
that compared the result to "varchar"[^1].

Now, the sized should be explicitly asked for. Adapt calls that needs
it.

[^1]: notably, the one that handle the conversion of the translated
fields to jsonb.

Part-of: #345
Related: odoo/upgrade#8715
Signed-off-by: Christophe Simonis (chs) <[email protected]>

Author: KangOl

src/base/tests/test_util.py

@@ -720,18 +720,21 @@ def test_iter_browse_call_twice(self):
 class TestPG(UnitTestCase):
     @parametrize(
         [
-            ("res_country", "name", "jsonb" if util.version_gte("16.0") else "varchar"),  # translated field
-            ("res_country", "code", "varchar(2)"),
-            ("res_currency", "active", "bool"),
-            ("res_country", "create_date", "timestamp"),
-            ("res_currency", "create_uid", "int4"),
-            ("res_country", "name_position", "varchar"),
-            ("res_country", "address_format", "text"),
-            ("res_partner", "does_not_exists", None),
+            ("res_country", "name", False, "jsonb" if util.version_gte("16.0") else "varchar"),  # translated field
+            ("res_country", "code", False, "varchar"),
+            ("res_country", "code", True, "varchar(2)"),
+            ("res_currency", "active", False, "bool"),
+            ("res_currency", "active", True, "bool"),
+            ("res_country", "create_date", False, "timestamp"),
+            ("res_currency", "create_uid", False, "int4"),
+            ("res_country", "name_position", False, "varchar"),
+            ("res_country", "name_position", True, "varchar"),
+            ("res_country", "address_format", False, "text"),
+            ("res_partner", "does_not_exists", False, None),
         ]
     )
-    def test_column_type(self, table, column, expected):
-        value = util.column_type(self.env.cr, table, column)
+    def test_column_type(self, table, column, sized, expected):
+        value = util.column_type(self.env.cr, table, column, sized=sized)
         if expected is None:
             self.assertIsNone(value)
         else:
@@ -762,11 +765,14 @@ def test_alter_column_type(self):
             "Some values where not casted correctly via USING",
         )
 
-        self.assertEqual(util.column_type(cr, "res_partner_bank", "y"), "varchar(4)")
+        self.assertEqual(util.column_type(cr, "res_partner_bank", "y"), "varchar")
+        self.assertEqual(util.column_type(cr, "res_partner_bank", "y", sized=True), "varchar(4)")
         util.alter_column_type(cr, "res_partner_bank", "y", "varchar")
         self.assertEqual(util.column_type(cr, "res_partner_bank", "y"), "varchar")
+        self.assertEqual(util.column_type(cr, "res_partner_bank", "y", sized=True), "varchar")
         util.alter_column_type(cr, "res_partner_bank", "y", "varchar(12)")
-        self.assertEqual(util.column_type(cr, "res_partner_bank", "y"), "varchar(12)")
+        self.assertEqual(util.column_type(cr, "res_partner_bank", "y"), "varchar")
+        self.assertEqual(util.column_type(cr, "res_partner_bank", "y", sized=True), "varchar(12)")
 
     @parametrize(
         [

src/util/pg.py

@@ -501,15 +501,15 @@ def get_value_or_en_translation(cr, table, column):
 
 
 def _column_info(cr, table, column):
+    # -> tuple[str, int | None, bool, bool] | None
     _validate_table(table)
-    # NOTE: usage of both `CONCAT` and `||` in the query below is done on purpose to take advantage of their NULL handling.
-    #       NULLS propagate with `||` and are ignored by `CONCAT`.
     cr.execute(
         """
-        SELECT CONCAT(
-                 COALESCE(bt.typname, t.typname),
-                 '(' || information_schema._pg_char_max_length(information_schema._pg_truetypid(a.*, t.*), information_schema._pg_truetypmod(a.*, t.*)) || ')'
-               ) AS udt_name,
+        SELECT COALESCE(bt.typname, t.typname) AS udt_name,
+               information_schema._pg_char_max_length(
+                    information_schema._pg_truetypid(a.*, t.*),
+                   information_schema._pg_truetypmod(a.*, t.*)
+               ) AS char_max_length,
                NOT (a.attnotnull OR t.typtype = 'd' AND t.typnotnull) AS is_nullable,
                (   c.relkind IN ('r','p','v','f')
                AND pg_column_is_updatable(c.oid::regclass, a.attnum, false)
@@ -589,7 +589,7 @@ def column_exists(cr, table, column):
     return _COLUMNS[(table, column)] if (table, column) in _COLUMNS else (_column_info(cr, table, column) is not None)
 
 
-def column_type(cr, table, column):
+def column_type(cr, table, column, sized=False):
     """
     Return the type of a column, if it exists.
 
@@ -598,17 +598,21 @@ def column_type(cr, table, column):
     :rtype: SQL type of the column
     """
     nfo = _column_info(cr, table, column)
+    if not nfo:
+        return None
+    if sized and nfo[1]:
+        return "{}({})".format(nfo[0], nfo[1])
     return nfo[0] if nfo else None
 
 
 def column_nullable(cr, table, column):
     nfo = _column_info(cr, table, column)
-    return nfo and nfo[1]
+    return nfo and nfo[2]
 
 
 def column_updatable(cr, table, column):
     nfo = _column_info(cr, table, column)
-    return nfo and nfo[2]
+    return nfo and nfo[3]
 
 
 def _normalize_pg_type(type_):
@@ -672,7 +676,7 @@ def create_column(cr, table, column, definition, **kwargs):
     if definition == "bool" and default is no_def:
         default = False
 
-    curtype = column_type(cr, table, column)
+    curtype = column_type(cr, table, column, sized=True)
     if curtype:
         if curtype != definition:
             _logger.error("%s.%s already exists but is %r instead of %r", table, column, curtype, definition)
@@ -738,7 +742,7 @@ def alter_column_type(cr, table, column, type, using=None, where=None, logger=_l
         raise ValueError("`where` parameter is only relevant with a non-default `using` parameter")
 
     if not using:
-        current_type = column_type(cr, table, column)
+        current_type = column_type(cr, table, column, sized=True)
         if current_type and current_type == _normalize_pg_type(type):
             logger.info("Column %r of table %r is already defined as %r", column, table, type)
             return
@@ -1884,7 +1888,7 @@ def bulk_update_table(cr, table, columns, mapping, key_col="id"):
         cols=ColumnList.from_unquoted(cr, columns),
         cols_values=SQLStr(
             ", ".join(
-                "(m.value->>{:d})::{}".format(col_idx, column_type(cr, table, col_name))
+                "(m.value->>{:d})::{}".format(col_idx, column_type(cr, table, col_name, sized=True))
                 for col_idx, col_name in enumerate(columns)
             )
         ),