def _is_field_compatible()

in pyiceberg/schema.py [0:0]

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

    def _is_field_compatible(self, lhs: NestedField) -> bool:
        # Validate nullability first.
        # An optional field can be missing in the provided schema
        # But a required field must exist as a required field
        try:
            rhs = self.provided_schema.find_field(lhs.field_id)
        except ValueError:
            if lhs.required:
                self.rich_table.add_row("❌", str(lhs), "Missing")
                return False
            else:
                self.rich_table.add_row("✅", str(lhs), "Missing")
                return True

        if lhs.required and not rhs.required:
            self.rich_table.add_row("❌", str(lhs), str(rhs))
            return False

        # Check type compatibility
        if lhs.field_type == rhs.field_type:
            self.rich_table.add_row("✅", str(lhs), str(rhs))
            return True
        # We only check that the parent node is also of the same type.
        # We check the type of the child nodes when we traverse them later.
        elif any(
            (isinstance(lhs.field_type, container_type) and isinstance(rhs.field_type, container_type))
            for container_type in {StructType, MapType, ListType}
        ):
            self.rich_table.add_row("✅", str(lhs), str(rhs))
            return True
        else:
            try:
                # If type can be promoted to the requested schema
                # it is considered compatible
                promote(rhs.field_type, lhs.field_type)
                self.rich_table.add_row("✅", str(lhs), str(rhs))
                return True
            except ResolveError:
                self.rich_table.add_row("❌", str(lhs), str(rhs))
                return False