docs: add type mapping tables between PyIceberg and PyArrow

[iamluan]

Closes#2226

Rationale for this change

This PR adds documentation with tables describing the type mapping between PyArrow and PyIceberg data types.

Are these changes tested?

Yes.
The changes are tested locally as shown in the image below.

Are there any user-facing changes?

Yes.
This PR adds new user-facing documentation.

[kevinjqliu]

this is great, thank you!
im not a big fan of documenting using python files. Could you add it as a markdown file instead? similar to #2480

Perhaps we can add it to the API section https://py.iceberg.apache.org/api/

[iamluan]

Thank you for your review. I have added the markdown to the API section.

[kevinjqliu]

LGTM! I added a few changes with the help of claude.
Mainly

• marked v3 Iceberg types
• expand eligible pyarrow types
• add numbered references to notes

[kevinjqliu]

Code reference:

iceberg-python/pyiceberg/io/pyarrow.py

Lines 697 to 833 in 44ce51a

	class_ConvertToArrowSchema(SchemaVisitorPerPrimitiveType[pa.DataType]):
	_metadata: dict[bytes, bytes]
	def__init__(
	self, metadata: dict[bytes, bytes] =EMPTY_DICT, include_field_ids: bool=True, file_format: FileFormat|None=None
	) ->None:
	self._metadata=metadata
	self._include_field_ids=include_field_ids
	self._file_format=file_format
	defschema(self, _: Schema, struct_result: pa.StructType) ->pa.schema:
	returnpa.schema(list(struct_result), metadata=self._metadata)
	defstruct(self, _: StructType, field_results: builtins.list[pa.DataType]) ->pa.DataType:
	returnpa.struct(field_results)
	deffield(self, field: NestedField, field_result: pa.DataType) ->pa.Field:
	metadata= {}
	iffield.doc:
	metadata[PYARROW_FIELD_DOC_KEY] =field.doc
	ifself._include_field_ids:
	# Add field ID based on file format
	ifself._file_format==FileFormat.ORC:
	metadata[ORC_FIELD_ID_KEY] =str(field.field_id)
	else:
	# Default to Parquet for backward compatibility
	metadata[PYARROW_PARQUET_FIELD_ID_KEY] =str(field.field_id)
	ifself._file_format==FileFormat.ORC:
	metadata[ORC_FIELD_REQUIRED_KEY] =str(field.required).lower()
	returnpa.field(
	name=field.name,
	type=field_result,
	nullable=field.optional,
	metadata=metadata,
	)
	deflist(self, list_type: ListType, element_result: pa.DataType) ->pa.DataType:
	element_field=self.field(list_type.element_field, element_result)
	returnpa.large_list(value_type=element_field)
	defmap(self, map_type: MapType, key_result: pa.DataType, value_result: pa.DataType) ->pa.DataType:
	key_field=self.field(map_type.key_field, key_result)
	value_field=self.field(map_type.value_field, value_result)
	returnpa.map_(key_type=key_field, item_type=value_field)
	defvisit_fixed(self, fixed_type: FixedType) ->pa.DataType:
	returnpa.binary(len(fixed_type))
	defvisit_decimal(self, decimal_type: DecimalType) ->pa.DataType:
	# It looks like decimal{32,64} is not fully implemented:
	# https://github.com/apache/arrow/issues/25483
	# https://github.com/apache/arrow/issues/43956
	# However, if we keep it as 128 in memory, and based on the
	# precision/scale Arrow will map it to INT{32,64}
	# https://github.com/apache/arrow/blob/598938711a8376cbfdceaf5c77ab0fd5057e6c02/cpp/src/parquet/arrow/schema.cc#L380-L392
	returnpa.decimal128(decimal_type.precision, decimal_type.scale)
	defvisit_boolean(self, _: BooleanType) ->pa.DataType:
	returnpa.bool_()
	defvisit_integer(self, _: IntegerType) ->pa.DataType:
	returnpa.int32()
	defvisit_long(self, _: LongType) ->pa.DataType:
	returnpa.int64()
	defvisit_float(self, _: FloatType) ->pa.DataType:
	# 32-bit IEEE 754 floating point
	returnpa.float32()
	defvisit_double(self, _: DoubleType) ->pa.DataType:
	# 64-bit IEEE 754 floating point
	returnpa.float64()
	defvisit_date(self, _: DateType) ->pa.DataType:
	# Date encoded as an int
	returnpa.date32()
	defvisit_time(self, _: TimeType) ->pa.DataType:
	returnpa.time64("us")
	defvisit_timestamp(self, _: TimestampType) ->pa.DataType:
	returnpa.timestamp(unit="us")
	defvisit_timestamp_ns(self, _: TimestampNanoType) ->pa.DataType:
	returnpa.timestamp(unit="ns")
	defvisit_timestamptz(self, _: TimestamptzType) ->pa.DataType:
	returnpa.timestamp(unit="us", tz="UTC")
	defvisit_timestamptz_ns(self, _: TimestamptzNanoType) ->pa.DataType:
	returnpa.timestamp(unit="ns", tz="UTC")
	defvisit_string(self, _: StringType) ->pa.DataType:
	returnpa.large_string()
	defvisit_uuid(self, _: UUIDType) ->pa.DataType:
	returnpa.uuid()
	defvisit_unknown(self, _: UnknownType) ->pa.DataType:
	"""Type `UnknownType` can be promoted to any primitive type in V3+ tables per the Iceberg spec."""
	returnpa.null()
	defvisit_binary(self, _: BinaryType) ->pa.DataType:
	returnpa.large_binary()
	defvisit_geometry(self, geometry_type: GeometryType) ->pa.DataType:
	"""Convert geometry type to PyArrow type.
	When geoarrow-pyarrow is available, returns a GeoArrow WKB extension type
	with CRS metadata. Otherwise, falls back to large_binary which stores WKB bytes.
	"""
	try:
	importgeoarrow.pyarrowasga
	returnga.wkb().with_crs(geometry_type.crs)
	exceptImportError:
	returnpa.large_binary()
	defvisit_geography(self, geography_type: GeographyType) ->pa.DataType:
	"""Convert geography type to PyArrow type.
	When geoarrow-pyarrow is available, returns a GeoArrow WKB extension type
	with CRS and edge type metadata. Otherwise, falls back to large_binary which stores WKB bytes.
	"""
	try:
	importgeoarrow.pyarrowasga
	wkb_type=ga.wkb().with_crs(geography_type.crs)
	# Map Iceberg algorithm to GeoArrow edge type
	ifgeography_type.algorithm=="spherical":
	wkb_type=wkb_type.with_edge_type(ga.EdgeType.SPHERICAL)
	# "planar" is the default edge type in GeoArrow, no need to set explicitly
	returnwkb_type
	exceptImportError:
	returnpa.large_binary()

[kevinjqliu]

Code reference:

iceberg-python/pyiceberg/io/pyarrow.py

Lines 1370 to 1513 in 44ce51a

	class_ConvertToIceberg(PyArrowSchemaVisitor[IcebergType|Schema]):
	"""Converts PyArrowSchema to Iceberg Schema. Applies the IDs from name_mapping if provided."""
	_field_names: builtins.list[str]
	def__init__(
	self, downcast_ns_timestamp_to_us: bool=False, format_version: TableVersion=TableProperties.DEFAULT_FORMAT_VERSION
	) ->None: # noqa: F821
	self._field_names= []
	self._downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us
	self._format_version=format_version
	def_field_id(self, field: pa.Field) ->int:
	if (field_id:=_get_field_id(field)) isnotNone:
	returnfield_id
	else:
	raiseValueError(f"Cannot convert {field} to Iceberg Field as field_id is empty.")
	defschema(self, schema: pa.Schema, struct_result: StructType) ->Schema:
	returnSchema(*struct_result.fields)
	defstruct(self, struct: pa.StructType, field_results: builtins.list[NestedField]) ->StructType:
	returnStructType(*field_results)
	deffield(self, field: pa.Field, field_result: IcebergType) ->NestedField:
	field_id=self._field_id(field)
	field_doc=doc_str.decode() if (field.metadataand (doc_str:=field.metadata.get(PYARROW_FIELD_DOC_KEY))) elseNone
	field_type=field_result
	returnNestedField(field_id, field.name, field_type, required=notfield.nullable, doc=field_doc)
	deflist(self, list_type: pa.ListType, element_result: IcebergType) ->ListType:
	element_field=list_type.value_field
	self._field_names.append(LIST_ELEMENT_NAME)
	element_id=self._field_id(element_field)
	self._field_names.pop()
	returnListType(element_id, element_result, element_required=notelement_field.nullable)
	defmap(self, map_type: pa.MapType, key_result: IcebergType, value_result: IcebergType) ->MapType:
	key_field=map_type.key_field
	self._field_names.append(MAP_KEY_NAME)
	key_id=self._field_id(key_field)
	self._field_names.pop()
	value_field=map_type.item_field
	self._field_names.append(MAP_VALUE_NAME)
	value_id=self._field_id(value_field)
	self._field_names.pop()
	returnMapType(key_id, key_result, value_id, value_result, value_required=notvalue_field.nullable)
	defprimitive(self, primitive: pa.DataType) ->PrimitiveType:
	ifpa.types.is_boolean(primitive):
	returnBooleanType()
	elifpa.types.is_integer(primitive):
	width=primitive.bit_width
	ifwidth<=32:
	returnIntegerType()
	elifwidth<=64:
	returnLongType()
	else:
	# Does not exist (yet)
	raiseTypeError(f"Unsupported integer type: {primitive}")
	elifpa.types.is_float32(primitive):
	returnFloatType()
	elifpa.types.is_float64(primitive):
	returnDoubleType()
	elifisinstance(primitive, pa.Decimal128Type):
	primitive=cast(pa.Decimal128Type, primitive)
	returnDecimalType(primitive.precision, primitive.scale)
	elifpa.types.is_string(primitive) orpa.types.is_large_string(primitive) orpa.types.is_string_view(primitive):
	returnStringType()
	elifpa.types.is_date32(primitive):
	returnDateType()
	elifisinstance(primitive, pa.Time64Type) andprimitive.unit=="us":
	returnTimeType()
	elifpa.types.is_timestamp(primitive):
	primitive=cast(pa.TimestampType, primitive)
	ifprimitive.unitin ("s", "ms", "us"):
	# Supported types, will be upcast automatically to 'us'
	pass
	elifprimitive.unit=="ns":
	ifself._downcast_ns_timestamp_to_us:
	logger.warning("Iceberg does not yet support 'ns' timestamp precision. Downcasting to 'us'.")
	elifself._format_version>=3:
	ifprimitive.tzinUTC_ALIASES:
	returnTimestamptzNanoType()
	elifprimitive.tzisNone:
	returnTimestampNanoType()
	else:
	raiseTypeError(
	"Iceberg does not yet support 'ns' timestamp precision. "
	"Use 'downcast-ns-timestamp-to-us-on-write' configuration property to automatically "
	"downcast 'ns' to 'us' on write.",
	)
	else:
	raiseTypeError(f"Unsupported precision for timestamp type: {primitive.unit}")
	ifprimitive.tzinUTC_ALIASES:
	returnTimestamptzType()
	elifprimitive.tzisNone:
	returnTimestampType()
	elifpa.types.is_binary(primitive) orpa.types.is_large_binary(primitive) orpa.types.is_binary_view(primitive):
	returnBinaryType()
	elifpa.types.is_fixed_size_binary(primitive):
	primitive=cast(pa.FixedSizeBinaryType, primitive)
	returnFixedType(primitive.byte_width)
	elifpa.types.is_null(primitive):
	# PyArrow null type (pa.null()) is converted to Iceberg UnknownType
	# UnknownType can be promoted to any primitive type in V3+ tables per the Iceberg spec
	ifself._format_version<3:
	field_path=".".join(self._field_names) ifself._field_nameselse"<root>"
	raiseValueError(
	"Null type (pa.null()) is not supported in Iceberg format version "
	f"{self._format_version}. Field: {field_path}. "
	"Requires format-version=3+ or use a concrete type (string, int, boolean, etc.)."
	)
	returnUnknownType()
	elifisinstance(primitive, pa.UuidType):
	returnUUIDType()
	raiseTypeError(f"Unsupported type: {primitive}")
	defbefore_field(self, field: pa.Field) ->None:
	self._field_names.append(field.name)
	defafter_field(self, field: pa.Field) ->None:
	self._field_names.pop()
	defbefore_list_element(self, element: pa.Field) ->None:
	self._field_names.append(LIST_ELEMENT_NAME)
	defafter_list_element(self, element: pa.Field) ->None:
	self._field_names.pop()
	defbefore_map_key(self, key: pa.Field) ->None:
	self._field_names.append(MAP_KEY_NAME)
	defafter_map_key(self, element: pa.Field) ->None:
	self._field_names.pop()
	defbefore_map_value(self, value: pa.Field) ->None:
	self._field_names.append(MAP_VALUE_NAME)
	defafter_map_value(self, element: pa.Field) ->None:
	self._field_names.pop()

[kevinjqliu]

Thanks for the PR @iamluan and thanks for the review @nssalian