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
class MappedField(IcebergBaseModel):
    field_id: int = Field(alias="field-id")
    names: List[str] = conlist(str, min_length=1)
    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 
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
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 
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."""
    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 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 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 __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]"

__len__()

Return the number of mappings.

Source code in pyiceberg/table/name_mapping.py 
84
85
86
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 
88
89
90
91
92
93
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[T], ABC

Source code in pyiceberg/table/name_mapping.py 
 99
100
101
102
103
104
105
106
107
108
109
110
class NameMappingVisitor(Generic[T], ABC):
    @abstractmethod
    def mapping(self, nm: NameMapping, field_results: T) -> T:
        """Visit a NameMapping."""

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

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

field(field, field_result) abstractmethod

Visit a MappedField.

Source code in pyiceberg/table/name_mapping.py 
108
109
110
@abstractmethod
def field(self, field: MappedField, field_result: T) -> T:
    """Visit a MappedField."""

fields(struct, field_results) abstractmethod

Visit a List[MappedField].

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

mapping(nm, field_results) abstractmethod

Visit a NameMapping.

Source code in pyiceberg/table/name_mapping.py 
100
101
102
@abstractmethod
def mapping(self, nm: NameMapping, field_results: T) -> T:
    """Visit a NameMapping."""

visit_name_mapping(obj, visitor)

Traverse the name mapping in post-order traversal.

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