o ˳iz^ @sFUdZddlZddlmZddlmZmZmZmZm Z m Z ddl m Z m Z mZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZm Z ddl!m"Z"e#e$Z%e e eeeeeeed Z&e'e(efe)d<eeee e d Z*e'e+e(e(fefe)d <e(eBZ,Gd d d Z-Gd ddee,Z.dS)z>Utility class for converting between Avro and Iceberg schemas.N)Any) FIELD_ID_PROPICEBERG_FIELD_NAME_PROPSchemaSchemaVisitorPerPrimitiveTypemake_compatible_namevisit) BinaryType BooleanTypeDateType DecimalType DoubleType FixedType FloatType IcebergType IntegerTypeListTypeLongTypeMapType NestedField PrimitiveType StringType StructType TimestampTypeTimestamptzTypeTimeType UnknownTypeUUIDType)decimal_required_bytes) booleanbytesdoublefloatintlongstringenumnullPRIMITIVE_FIELD_TYPE_MAPPING))dater#) time-microsr$)timestamp-microsr$)uuidfixed)r,r%LOGICAL_FIELD_TYPE_MAPPINGc@sveZdZdeeefdefddZd%dededBdefdd Z d eeefe eeeefBBeBde eeeefBe ffd d Z d eeeefBdefddZdeeefdefddZdeeefdefddZdeeefdefddZdeeefdefddZdeeefdefddZd eeefdefdd Zd eeefdefd!d"Zd eeefdefd#d$ZdS)&AvroSchemaConversion avro_schemareturncs tfdd|dDddiS)aConvert an Apache Avro into an Apache Iceberg schema equivalent. This expects to have field id's to be encoded in the Avro schema: { "type": "record", "name": "manifest_file", "fields": [ {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500}, {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501} ] } Example: This converts an Avro schema into an Iceberg schema: >>> avro_schema = AvroSchemaConversion().avro_to_iceberg({ ... "type": "record", ... "name": "manifest_file", ... "fields": [ ... {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500}, ... {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501} ... ] ... }) >>> iceberg_schema = Schema( ... NestedField( ... field_id=500, name="manifest_path", field_type=StringType(), ... required=False, doc="Location URI with FS scheme" ... ), ... NestedField( ... field_id=501, name="manifest_length", field_type=LongType(), ... required=False, doc="Total file size in bytes" ... ), ... schema_id=1 ... ) >>> avro_schema == iceberg_schema True Args: avro_schema (Dict[str, Any]): The JSON decoded Avro schema. Returns: Equivalent Iceberg schema. cg|]}|qS_convert_field.0fieldselfr3U/home/ubuntu/.local/lib/python3.10/site-packages/pyiceberg/utils/schema_conversion.py z8AvroSchemaConversion.avro_to_iceberg..fields schema_id)r)r:r0r3r9r;avro_to_icebergSs -z$AvroSchemaConversion.avro_to_icebergNschema schema_namecCst|t|S)zQConvert an Iceberg schema into an Avro dictionary that can be serialized to JSON.)rConvertSchemaToAvro)r:rBrCr3r3r;iceberg_to_avrosz$AvroSchemaConversion.iceberg_to_avro type_unioncCspt|tr |dfSt|tr|dfS|}t|dkr!td|d|dkr+tdttdd|dd fS) a' Convert Unions into their type and resolves if the field is required. Examples: >>> AvroSchemaConversion()._resolve_union('str') ('str', True) >>> AvroSchemaConversion()._resolve_union(['null', 'str']) ('str', False) >>> AvroSchemaConversion()._resolve_union([{'type': 'str'}]) ({'type': 'str'}, True) >>> AvroSchemaConversion()._resolve_union(['null', {'type': 'str'}]) ({'type': 'str'}, False) Args: type_union: The field, can be a string 'str', list ['null', 'str'], or dict {"type": 'str'}. Returns: A tuple containing the type and if required. Raises: TypeError: In the case non-optional union types are encountered. Tz=Non-optional types aren't part of the Iceberg specification: r'rzOnly null-unions are supportedcSs|dkSNr'r3)tr3r3r;sz5AvroSchemaConversion._resolve_union..F) isinstancestrdictlen TypeErrorlistfilter)r:rF avro_typesr3r3r;_resolve_unions    z#AvroSchemaConversion._resolve_union avro_typecCst|tr |tvr t|St|trqd|vr||Sd|vr5t|dtr5|d}d|vr5t|dts&|d}|dkrB||S|dkrK||S|dkrT||S|dkr]||St|trj|tvrjt|St d|t d|)z Resolve the Avro type. Args: avro_type: The Avro type, can be simple or complex. Returns: The equivalent IcebergType. Raises: ValueError: When there are unknown types logicalTypetyperecordarraymapr-zType not recognized: ) rKrLr(rM_convert_logical_type_convert_record_type_convert_array_type_convert_map_type_convert_fixed_typerO)r:rTtype_identifierr3r3r;_convert_schemas*       z$AvroSchemaConversion._convert_schemar8cCsTt|vrtdtd|||d\}}t|t|d||||ddS)zConvert an Avro field into an Iceberg equivalent field. Args: field: The Avro field. Returns: The Iceberg equivalent field. zCannot convert field, missing z: rVnamedoc)field_idra field_typerequiredrb)r ValueErrorrSrr`get)r:r8 plain_typerer3r3r;r5s z#AvroSchemaConversion._convert_field record_typecs4|ddkr td|tfdd|dDS)aE Convert the fields from a record into an Iceberg struct. Examples: >>> from pyiceberg.utils.schema_conversion import AvroSchemaConversion >>> record_type = { ... "type": "record", ... "name": "r508", ... "fields": [{ ... "name": "contains_null", ... "type": "boolean", ... "doc": "True if any file has a null partition value", ... "field-id": 509, ... }, { ... "name": "contains_nan", ... "type": ["null", "boolean"], ... "doc": "True if any file has a nan partition value", ... "default": None, ... "field-id": 518, ... }], ... } >>> actual = AvroSchemaConversion()._convert_record_type(record_type) >>> expected = StructType( ... fields=( ... NestedField( ... field_id=509, ... name="contains_null", ... field_type=BooleanType(), ... required=False, ... doc="True if any file has a null partition value", ... ), ... NestedField( ... field_id=518, ... name="contains_nan", ... field_type=BooleanType(), ... required=True, ... doc="True if any file has a nan partition value", ... ), ... ) ... ) >>> expected == actual True Args: record_type: The record type itself. Returns: A StructType. rVrWzExpected record type, got: cr2r3r4r6r9r3r;r<'r=z=AvroSchemaConversion._convert_record_type..r>)rfr)r:rir3r9r;r[s 1z)AvroSchemaConversion._convert_record_type array_typecCs@d|vr td|||d\}}t|d|||dS)N element-idz/Cannot convert array-type, missing element-id: items) element_id element_typeelement_required)rfrSrr`)r:rjrhror3r3r;r\)sz(AvroSchemaConversion._convert_array_typemap_typecCs4||d\}}t|dt|d|||dS)aQConvert an avro map type into an Iceberg MapType. Args: map_type: The dict that describes the Avro map type. Examples: >>> from pyiceberg.utils.schema_conversion import AvroSchemaConversion >>> avro_field = { ... "type": "map", ... "values": ["null", "long"], ... "key-id": 101, ... "value-id": 102, ... } >>> actual = AvroSchemaConversion()._convert_map_type(avro_field) >>> expected = MapType( ... key_id=101, ... key_type=StringType(), ... value_id=102, ... value_type=LongType(), ... value_required=True ... ) >>> actual == expected True Returns: A MapType. valueskey-idvalue-idkey_idkey_typevalue_id value_typevalue_required)rSrrr`)r:rprxryr3r3r;r]5sz&AvroSchemaConversion._convert_map_typeavro_logical_typecCs~|d}|d}|dkr||S|dkr||S|dkr,|dddur)tStS||ftvr8t||fStd |) aConvert a schema with a logical type annotation into an IcebergType. For the decimal and map we need to fetch more keys from the dict, and for the simple ones we can just look it up in the mapping. Examples: >>> from pyiceberg.utils.schema_conversion import AvroSchemaConversion >>> avro_logical_type = { ... "type": "int", ... "logicalType": "date" ... } >>> actual = AvroSchemaConversion()._convert_logical_type(avro_logical_type) >>> actual == DateType() True Args: avro_logical_type: The logical type. Returns: The converted logical type. Raises: ValueError: When the logical type is unknown. rUrVdecimalrYr+ adjust-to-utcFTz+Unknown logical/physical type combination: )_convert_logical_decimal_type_convert_logical_map_typergrrr.rf)r:rz logical_type physical_typer3r3r;rZZs    z*AvroSchemaConversion._convert_logical_typecCst|d|ddS)aConvert an avro type to an Iceberg DecimalType. Args: avro_type: The Avro type. Examples: >>> from pyiceberg.utils.schema_conversion import AvroSchemaConversion >>> avro_decimal_type = { ... "type": "bytes", ... "logicalType": "decimal", ... "precision": 19, ... "scale": 25 ... } >>> actual = AvroSchemaConversion()._convert_logical_decimal_type(avro_decimal_type) >>> expected = DecimalType( ... precision=19, ... scale=25 ... ) >>> actual == expected True Returns: A Iceberg DecimalType. precisionscale)rr)r r:rTr3r3r;r}sz2AvroSchemaConversion._convert_logical_decimal_typecCs~|dd}t|dkrtd|d|ttdd|d}|ttdd|d}t|j|j|j|j|jd S) asConvert an avro map type to an Iceberg MapType. In the case where a map hasn't a key as a type you can use a logical map to still encode this in Avro. Args: avro_type: The Avro Type. Examples: >>> from pyiceberg.utils.schema_conversion import AvroSchemaConversion >>> avro_type = { ... "type": "array", ... "logicalType": "map", ... "items": { ... "type": "record", ... "name": "k101_v102", ... "fields": [ ... {"name": "key", "type": "int", "field-id": 101}, ... {"name": "value", "type": "string", "field-id": 102}, ... ], ... }, ... } >>> actual = AvroSchemaConversion()._convert_logical_map_type(avro_type) >>> expected = MapType( ... key_id=101, ... key_type=IntegerType(), ... value_id=102, ... value_type=StringType(), ... value_required=False ... ) >>> actual == expected True .. _Apache Iceberg specification: https://iceberg.apache.org/spec/#appendix-a-format-specific-requirements Returns: The logical map. rlr>rGzInvalid key-value pair schema: cS |ddkS)Nrakeyr3fr3r3r;rJ z@AvroSchemaConversion._convert_logical_map_type..rcSr)Nravaluer3rr3r3r;rJrrt) rNrfr5rPrQrrcrdre)r:rTr>rrr3r3r;r~s ' z.AvroSchemaConversion._convert_logical_map_typecCst|ddS)av Convert Avro Type to the equivalent Iceberg fixed type. - https://avro.apache.org/docs/current/spec.html#Fixed Args: avro_type: The Avro type. Examples: >>> from pyiceberg.utils.schema_conversion import AvroSchemaConversion >>> avro_fixed_type = { ... "name": "md5", ... "type": "fixed", ... "size": 16 ... } >>> FixedType(length=16) == AvroSchemaConversion()._convert_fixed_type(avro_fixed_type) True Returns: An Iceberg equivalent fixed type. size)length)rrr3r3r;r^sz(AvroSchemaConversion._convert_fixed_typeN)__name__ __module__ __qualname__rMrLrrrAAvroTyperErPtupleboolrSrr`rr5rr[rr\rr]rZr r}r~rr^r3r3r3r;r/Rs/F1&6 %)4r/c@s*eZdZUdZedBed<eed<eed<eed<dedBddfdd Zd ed e de fd d Z de ddfddZ de ddfddZ de ddfddZdedee de fddZde de de fddZded e de fd!d"Zd#ed$e d%e de fd&d'Zd(ede fd)d*Zd+ede fd,d-Zd.ede fd/d0Zd1ede fd2d3Zd4ede fd5d6Zd7e de fd8d9Z!d:e"de fd;d<Z#d=e$de fd>d?Z%d@e&de fdAdBZ'dCe(de fdDdEZ)dCe(de fdFdGZ*dHe+de fdIdJZ,dHe+de fdKdLZ-dMe.de fdNdOZ/dPe0de fdQdRZ1dSe2de fdTdUZ3dVe4de fdWdXZ5dS)YrDz,Convert an Iceberg schema to an Avro schema.NrClast_list_field_idlast_map_key_field_idlast_map_value_field_idr1cCs ||_dS)zzConvert an Iceberg schema to an Avro schema. Args: schema_name: The name of the root record. N)rC)r:rCr3r3r;__init__s zConvertSchemaToAvro.__init__rB struct_resultcCs"t|tr|jdur|j|d<|S)Nra)rKrMrC)r:rBrr3r3r;rBs zConvertSchemaToAvro.schemaelementcC |j|_dSr)rcr)r:rr3r3r;before_list_element z'ConvertSchemaToAvro.before_list_elementrcCrr)rcr)r:rr3r3r;before_map_keyrz"ConvertSchemaToAvro.before_map_keyrcCrr)rcr)r:rr3r3r;before_map_valuerz$ConvertSchemaToAvro.before_map_valuestruct field_resultscCs d|dS)NrW)rVr>r3)r:rrr3r3r;r  zConvertSchemaToAvro.structr8 field_resultcCst|tr|ddkrd|j|d<|j}t|}d|t|jd|jr&|nd|gi}||kr3||t<|j dur>|j |d<n|j rEd|d<|j durO|j |d<|S)NrVrWrrar'defaultrb) rKrMrgrcrarrrer write_defaultoptionalrb)r:r8r original_namesanitized_nameresultr3r3r;r8 s"    zConvertSchemaToAvro.field list_typeelement_resultcCs6t|tr|ddkrd|j|d<d|j|dS)NrVrWrrarX)rVrkrl)rKrMrgr)r:rrr3r3r;rP'szConvertSchemaToAvro.listrp key_result value_resultc Cs`t|trd||j|jdSddd|jd|jddd |t|jidd d |t|jigd dd S) NrY)rVrqrrrsrXrWk_vrarrVr)rVrar>)rVrlrU)rKrrrr)r:rprrr3r3r;rY-s  zConvertSchemaToAvro.map fixed_typecCsdt|dt|dS)Nr-fixed_)rVrra)rN)r:rr3r3r; visit_fixedFszConvertSchemaToAvro.visit_fixed decimal_typec Cs,dt|jd|j|jd|jd|jdS)Nr-r{decimal__)rVrrUrrra)rrr)r:rr3r3r; visit_decimalIsz!ConvertSchemaToAvro.visit_decimal boolean_typecCdS)Nrr3)r:rr3r3r; visit_booleanSz!ConvertSchemaToAvro.visit_boolean integer_typecCr)Nr#r3)r:rr3r3r; visit_integerVrz!ConvertSchemaToAvro.visit_integer long_typecCr)Nr$r3)r:rr3r3r; visit_longYrzConvertSchemaToAvro.visit_long float_typecCr)Nr"r3)r:rr3r3r; visit_float\rzConvertSchemaToAvro.visit_float double_typecCr)Nr!r3)r:rr3r3r; visit_double_rz ConvertSchemaToAvro.visit_double date_typecC dddS)Nr#r)rVrUr3)r:rr3r3r; visit_datebrzConvertSchemaToAvro.visit_date time_typecCr)Nr$r*rr3)r:rr3r3r; visit_timeerzConvertSchemaToAvro.visit_timetimestamp_typecC ddddS)Nr$r+FrVrUr|r3r:rr3r3r;visit_timestamphrz#ConvertSchemaToAvro.visit_timestampcCr)Nr$timestamp-nanosFrr3rr3r3r;visit_timestamp_nskrz&ConvertSchemaToAvro.visit_timestamp_nstimestamptz_typecCr)Nr$r+Trr3r:rr3r3r;visit_timestamptznrz%ConvertSchemaToAvro.visit_timestamptzcCr)Nr$rTrr3rr3r3r;visit_timestamptz_nsqrz(ConvertSchemaToAvro.visit_timestamptz_ns string_typecCr)Nr%r3)r:rr3r3r; visit_stringtrz ConvertSchemaToAvro.visit_string uuid_typecCsdddddS)Nr-r, uuid_fixed)rVrrUrar3)r:rr3r3r; visit_uuidwszConvertSchemaToAvro.visit_uuid binary_typecCr)Nr r3)r:rr3r3r; visit_binaryzrz ConvertSchemaToAvro.visit_binary unknown_typecCrrHr3)r:rr3r3r; visit_unknown}rz!ConvertSchemaToAvro.visit_unknown)6rrr__doc__rL__annotations__r#rrrrBrrrrrrPrr8rrrYrrr rr rrrrrrrr rr rrrrrrrrrrrrrr rrrr3r3r3r;rDs@   rD)/rloggingtypingrpyiceberg.schemarrrrrrpyiceberg.typesr r r r r rrrrrrrrrrrrrrrrpyiceberg.utils.decimalr getLoggerrloggerr(rMrLrr.rrr/rDr3r3r3r;s:  \