Skip to content

name_mapping

Contains everything around the name mapping.

More information can be found on here: https://iceberg.apache.org/spec/#name-mapping-serialization

MappedField

Bases: IcebergBaseModel

Source code in pyiceberg/table/name_mapping.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class MappedField(IcebergBaseModel):
    field_id: int = Field(alias="field-id")
    names: List[str] = conlist(str)
    fields: List[MappedField] = Field(default_factory=list)

    @field_validator("fields", mode="before")
    @classmethod
    def convert_null_to_empty_List(cls, v: Any) -> Any:
        return v or []

    @model_serializer
    def ser_model(self) -> Dict[str, Any]:
        """Set custom serializer to leave out the field when it is empty."""
        fields = {"fields": self.fields} if len(self.fields) > 0 else {}
        return {
            "field-id": self.field_id,
            "names": self.names,
            **fields,
        }

    def __len__(self) -> int:
        """Return the number of fields."""
        return len(self.fields)

    def __str__(self) -> str:
        """Convert the mapped-field into a nicely formatted string."""
        # Otherwise the UTs fail because the order of the set can change
        fields_str = ", ".join([str(e) for e in self.fields]) or ""
        fields_str = " " + fields_str if fields_str else ""
        return "([" + ", ".join(self.names) + "] -> " + (str(self.field_id) or "?") + fields_str + ")"

__len__()

Return the number of fields.

Source code in pyiceberg/table/name_mapping.py 
59
60
61
def __len__(self) -> int:
    """Return the number of fields."""
    return len(self.fields)

__str__()

Convert the mapped-field into a nicely formatted string.

Source code in pyiceberg/table/name_mapping.py 
63
64
65
66
67
68
def __str__(self) -> str:
    """Convert the mapped-field into a nicely formatted string."""
    # Otherwise the UTs fail because the order of the set can change
    fields_str = ", ".join([str(e) for e in self.fields]) or ""
    fields_str = " " + fields_str if fields_str else ""
    return "([" + ", ".join(self.names) + "] -> " + (str(self.field_id) or "?") + fields_str + ")"

ser_model()

Set custom serializer to leave out the field when it is empty.

Source code in pyiceberg/table/name_mapping.py 
49
50
51
52
53
54
55
56
57
@model_serializer
def ser_model(self) -> Dict[str, Any]:
    """Set custom serializer to leave out the field when it is empty."""
    fields = {"fields": self.fields} if len(self.fields) > 0 else {}
    return {
        "field-id": self.field_id,
        "names": self.names,
        **fields,
    }

NameMapping

Bases: IcebergRootModel[List[MappedField]]

Source code in pyiceberg/table/name_mapping.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class NameMapping(IcebergRootModel[List[MappedField]]):
    root: List[MappedField]

    @cached_property
    def _field_by_name(self) -> Dict[str, MappedField]:
        return visit_name_mapping(self, _IndexByName())

    @deprecated(
        deprecated_in="0.8.0",
        removed_in="0.9.0",
        help_message="Please use `apply_name_mapping` instead",
    )
    def find(self, *names: str) -> MappedField:
        name = ".".join(names)
        try:
            return self._field_by_name[name]
        except KeyError as e:
            raise ValueError(f"Could not find field with name: {name}") from e

    def __len__(self) -> int:
        """Return the number of mappings."""
        return len(self.root)

    def __iter__(self) -> Iterator[MappedField]:
        """Iterate over the mapped fields."""
        return iter(self.root)

    def __str__(self) -> str:
        """Convert the name-mapping into a nicely formatted string."""
        if len(self.root) == 0:
            return "[]"
        else:
            return "[\n  " + "\n  ".join([str(e) for e in self.root]) + "\n]"

__iter__()

Iterate over the mapped fields.

Source code in pyiceberg/table/name_mapping.py 
94
95
96
def __iter__(self) -> Iterator[MappedField]:
    """Iterate over the mapped fields."""
    return iter(self.root)

__len__()

Return the number of mappings.

Source code in pyiceberg/table/name_mapping.py 
90
91
92
def __len__(self) -> int:
    """Return the number of mappings."""
    return len(self.root)

__str__()

Convert the name-mapping into a nicely formatted string.

Source code in pyiceberg/table/name_mapping.py 
 98
 99
100
101
102
103
def __str__(self) -> str:
    """Convert the name-mapping into a nicely formatted string."""
    if len(self.root) == 0:
        return "[]"
    else:
        return "[\n  " + "\n  ".join([str(e) for e in self.root]) + "\n]"

NameMappingVisitor

Bases: Generic[S, T], ABC

Source code in pyiceberg/table/name_mapping.py 
110
111
112
113
114
115
116
117
118
119
120
121
class NameMappingVisitor(Generic[S, T], ABC):
    @abstractmethod
    def mapping(self, nm: NameMapping, field_results: S) -> S:
        """Visit a NameMapping."""

    @abstractmethod
    def fields(self, struct: List[MappedField], field_results: List[T]) -> S:
        """Visit a List[MappedField]."""

    @abstractmethod
    def field(self, field: MappedField, field_result: S) -> T:
        """Visit a MappedField."""

field(field, field_result) abstractmethod

Visit a MappedField.

Source code in pyiceberg/table/name_mapping.py 
119
120
121
@abstractmethod
def field(self, field: MappedField, field_result: S) -> T:
    """Visit a MappedField."""

fields(struct, field_results) abstractmethod

Visit a List[MappedField].

Source code in pyiceberg/table/name_mapping.py 
115
116
117
@abstractmethod
def fields(self, struct: List[MappedField], field_results: List[T]) -> S:
    """Visit a List[MappedField]."""

mapping(nm, field_results) abstractmethod

Visit a NameMapping.

Source code in pyiceberg/table/name_mapping.py 
111
112
113
@abstractmethod
def mapping(self, nm: NameMapping, field_results: S) -> S:
    """Visit a NameMapping."""

visit_name_mapping(obj, visitor)

Traverse the name mapping in post-order traversal.

Source code in pyiceberg/table/name_mapping.py 
142
143
144
145
@singledispatch
def visit_name_mapping(obj: Union[NameMapping, List[MappedField], MappedField], visitor: NameMappingVisitor[S, T]) -> S:
    """Traverse the name mapping in post-order traversal."""
    raise NotImplementedError(f"Cannot visit non-type: {obj}")