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 
38
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: Optional[int] = Field(alias="field-id", default=None)
    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."""
        serialized: Dict[str, Any] = {"names": self.names}
        if self.field_id is not None:
            serialized["field-id"] = self.field_id
        if len(self.fields) > 0:
            serialized["fields"] = self.fields
        return serialized

    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 ""
        field_id = "?" if self.field_id is None else (str(self.field_id) or "?")
        return "([" + ", ".join(self.names) + "] -> " + field_id + fields_str + ")"

__len__()

Return the number of fields.

Source code in pyiceberg/table/name_mapping.py 
58
59
60
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 
62
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 ""
    field_id = "?" if self.field_id is None else (str(self.field_id) or "?")
    return "([" + ", ".join(self.names) + "] -> " + field_id + fields_str + ")"

ser_model()

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

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

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
class NameMapping(IcebergRootModel[List[MappedField]]):
    root: List[MappedField]

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

    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 
82
83
84
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 
78
79
80
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 
86
87
88
89
90
91
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 
 98
 99
100
101
102
103
104
105
106
107
108
109
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 
107
108
109
@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 
103
104
105
@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 
 99
100
101
@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 
130
131
132
133
@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}")