0.0.198

Author: joocer

orso/types.py

@@ -15,6 +15,7 @@
 import re
 from enum import Enum
 from typing import Any
+from typing import Iterable
 from typing import Tuple
 from typing import Type
 from typing import Union
@@ -95,7 +96,7 @@ def __init__(self, *args, **kwargs):
 
     def is_numeric(self):
         """is the typle number-based"""
-        return self in (self.INTEGER, self.DOUBLE, self.DECIMAL)
+        return self in (self.INTEGER, self.DOUBLE, self.DECIMAL, self.BOOLEAN)
 
     def is_temporal(self):
         """is the type time-based"""
@@ -222,3 +223,59 @@ def from_name(name: str) -> tuple:
     OrsoTypes.JSONB: bytes,
     OrsoTypes.NULL: lambda x: None,
 }
+
+
+def find_compatible_type(types: Iterable[OrsoTypes]) -> OrsoTypes:
+    """
+    Find the most compatible type that can represent all input types.
+
+    Parameters:
+        types (list): List of OrsoTypes to find a compatible type for
+
+    Returns:
+        OrsoTypes: The most compatible type that can represent all input types
+
+    Examples:
+        >>> OrsoTypes.find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.DOUBLE])
+        OrsoTypes.DOUBLE
+        >>> OrsoTypes.find_compatible_type([OrsoTypes.BLOB, OrsoTypes.VARCHAR])
+        OrsoTypes.VARCHAR
+    """
+    if not types:
+        return OrsoTypes.NULL
+
+    # Handle single type case
+    if len(set(types)) == 1:
+        return types[0]
+
+    # Define type promotion hierarchy
+    type_hierarchy = {
+        # Numeric promotion
+        OrsoTypes.BOOLEAN: 1,
+        OrsoTypes.INTEGER: 2,
+        OrsoTypes.DOUBLE: 3,
+        OrsoTypes.DECIMAL: 4,
+        # Temporal promotion
+        OrsoTypes.DATE: 1,
+        OrsoTypes.TIMESTAMP: 2,
+        # String/binary promotion
+        OrsoTypes.BLOB: 1,
+        OrsoTypes.VARCHAR: 2,
+    }
+
+    # First check if all types are in the same category
+    if all(t.is_numeric() for t in types):
+        return max(types, key=lambda t: type_hierarchy.get(t, 0))
+    if all(t.is_temporal() for t in types):
+        return max(types, key=lambda t: type_hierarchy.get(t, 0))
+    if all(t.is_large_object() for t in types):
+        return max(types, key=lambda t: type_hierarchy.get(t, 0))
+    if all(
+        t in (OrsoTypes.BLOB, OrsoTypes.STRUCT, OrsoTypes.JSONB, OrsoTypes.VARCHAR) for t in types
+    ):
+        return OrsoTypes.BLOB
+
+    # For heterogeneous types, default to the most flexible type
+    if any(t == OrsoTypes.BLOB for t in types):
+        return OrsoTypes.BLOB
+    return OrsoTypes.VARCHAR

orso/version.py

@@ -10,5 +10,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__: str = "0.0.197"
+__version__: str = "0.0.198"
 __author__: str = "@joocer"

tests/test_schema.py

@@ -409,6 +409,44 @@ def test_type_name_parser():
     (_type, _length, _scale, _precision, _element_type) = OrsoTypes.from_name("ARRAY<TIMESTAMP>")
     assert (_type, _length, _scale, _precision, _element_type) == (OrsoTypes.ARRAY, None, None, None, OrsoTypes.TIMESTAMP)
 
+def test_type_combinations():
+
+    from orso.types import find_compatible_type
+
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.INTEGER]) == OrsoTypes.INTEGER
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.DOUBLE]) == OrsoTypes.DOUBLE
+    assert find_compatible_type([OrsoTypes.DOUBLE, OrsoTypes.INTEGER]) == OrsoTypes.DOUBLE
+    assert find_compatible_type([OrsoTypes.DOUBLE, OrsoTypes.DOUBLE]) == OrsoTypes.DOUBLE
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.DECIMAL]) == OrsoTypes.DECIMAL
+    assert find_compatible_type([OrsoTypes.DECIMAL, OrsoTypes.INTEGER]) == OrsoTypes.DECIMAL
+    assert find_compatible_type([OrsoTypes.DECIMAL, OrsoTypes.DECIMAL]) == OrsoTypes.DECIMAL
+    assert find_compatible_type([OrsoTypes.DECIMAL, OrsoTypes.DOUBLE]) == OrsoTypes.DECIMAL
+    assert find_compatible_type([OrsoTypes.DOUBLE, OrsoTypes.DECIMAL]) == OrsoTypes.DECIMAL
+    assert find_compatible_type([OrsoTypes.DOUBLE, OrsoTypes.DOUBLE]) == OrsoTypes.DOUBLE
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.TIMESTAMP]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.TIMESTAMP, OrsoTypes.INTEGER]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.TIMESTAMP, OrsoTypes.TIMESTAMP]) == OrsoTypes.TIMESTAMP
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.VARCHAR]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.VARCHAR, OrsoTypes.INTEGER]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.VARCHAR, OrsoTypes.VARCHAR]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.BLOB]) == OrsoTypes.BLOB
+    assert find_compatible_type([OrsoTypes.BLOB, OrsoTypes.INTEGER]) == OrsoTypes.BLOB
+    assert find_compatible_type([OrsoTypes.BLOB, OrsoTypes.BLOB]) == OrsoTypes.BLOB
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.BOOLEAN]) == OrsoTypes.INTEGER
+    assert find_compatible_type([OrsoTypes.BOOLEAN, OrsoTypes.INTEGER]) == OrsoTypes.INTEGER
+    assert find_compatible_type([OrsoTypes.BOOLEAN, OrsoTypes.BOOLEAN]) == OrsoTypes.BOOLEAN
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.DATE]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.DATE, OrsoTypes.INTEGER]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.DATE, OrsoTypes.DATE]) == OrsoTypes.DATE
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.ARRAY]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.ARRAY, OrsoTypes.INTEGER]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.ARRAY, OrsoTypes.ARRAY]) == OrsoTypes.ARRAY
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.STRUCT]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.STRUCT, OrsoTypes.INTEGER]) == OrsoTypes.VARCHAR
+    assert find_compatible_type([OrsoTypes.STRUCT, OrsoTypes.STRUCT]) == OrsoTypes.STRUCT
+    assert find_compatible_type([OrsoTypes.INTEGER, OrsoTypes.JSONB]) == OrsoTypes.VARCHAR
+
+
 if __name__ == "__main__":  # prgama: nocover
     from tests import run_tests