add UnknownTransform

python/src/iceberg/transforms.py

@@ -222,6 +222,33 @@ def hash(self, value: UUID) -> int:
         )
 
 
+class UnknownTransform(Transform):
+    """A transform that represents when an unknown transform is provided
+    Args:
+      source_type (Type): An Iceberg `Type`
+      transform (str): A string name of a transform
+    Raises:
+      AttributeError: If the apply method is called.
+    """
+
+    def __init__(self, source_type: IcebergType, transform: str):
+        super().__init__(
+            transform,
+            f"transforms.UnknownTransform(source_type={repr(source_type)}, transform={repr(transform)})",
+        )
+        self._type = source_type
+        self._transform = transform
+
+    def apply(self, value):
+        raise AttributeError(f"Cannot apply unsupported transform: {self}")
+
+    def can_transform(self, target: IcebergType) -> bool:
+        return self._type == target
+
+    def result_type(self, source: IcebergType) -> IcebergType:
+        return StringType()
+
+
 def bucket(source_type: IcebergType, num_buckets: int) -> BaseBucketTransform:
     if type(source_type) in {IntegerType, LongType, DateType, TimeType, TimestampType, TimestamptzType}:
         return BucketNumberTransform(source_type, num_buckets)

python/tests/test_transforms.py

@@ -24,6 +24,7 @@
 from iceberg import transforms
 from iceberg.types import (
     BinaryType,
+    BooleanType,
     DateType,
     DecimalType,
     FixedType,
@@ -125,3 +126,13 @@ def test_string_with_surrogate_pair():
     as_bytes = bytes(string_with_surrogate_pair, "UTF-8")
     bucket_transform = transforms.bucket(StringType(), 100)
     assert bucket_transform.hash(string_with_surrogate_pair) == mmh3.hash(as_bytes)
+
+
+def test_unknown_transform():
+    unknown_transform = transforms.UnknownTransform(FixedType(8), "unknown")
+    assert str(unknown_transform) == str(eval(repr(unknown_transform)))
+    with pytest.raises(AttributeError):
+        unknown_transform.apply("test")
+    assert unknown_transform.can_transform(FixedType(8))
+    assert not unknown_transform.can_transform(FixedType(5))
+    assert isinstance(unknown_transform.result_type(BooleanType()), StringType)