def _partition_field()

in pyiceberg/table/update/spec.py [0:0]

• 16 lines of code
• 10 McCabe index (conditional complexity)

    def _partition_field(self, transform_key: Tuple[int, Transform[Any, Any]], name: Optional[str]) -> PartitionField:
        if self._transaction.table_metadata.format_version == 2:
            source_id, transform = transform_key
            historical_fields = []
            for spec in self._transaction.table_metadata.specs().values():
                for field in spec.fields:
                    historical_fields.append(field)

            for field in historical_fields:
                if field.source_id == source_id and repr(field.transform) == repr(transform):
                    if name is None or field.name == name:
                        return PartitionField(source_id, field.field_id, transform, field.name)

        new_field_id = self._new_field_id()
        if name is None:
            tmp_field = PartitionField(transform_key[0], new_field_id, transform_key[1], "unassigned_field_name")
            name = _visit_partition_field(self._transaction.table_metadata.schema(), tmp_field, _PartitionNameGenerator())
        return PartitionField(transform_key[0], new_field_id, transform_key[1], name)