o `۷io @sddlZddlZddlZddlmZddlmZmZmZm Z m Z ddl m Z er,ddl Z dZdZeeZeaddZd d Zd d Zd dZddde edgdfe effddZdeddfddZdkddZdee eddfddddfdd Zd!d"de ed#dffd$d%Zd&ed#d'ddd"fd(d)ZeGd*d#d#Z dld,d-Z!dmd/d0Z"d'dde#fd1d2Z$d'dde#fd3d4Z%dnd6d7Z&dmd8d9Z'dmd:d;Z(dmd  #dod?d@Z*dpdBdCZ+dqdEdFZ, . G  #drdHdIZ-dsdKdLZ. . M  #dtdNdOZ/ddPd'e ddQe0dRe0ddPf dSdTZ1ddPdUe0dQe0dRe0ddPf dVdWZ2ddPdQe0dRe0ddPfdXdYZ3ddPdZddQe0dRe0de dPe0e0ff d[d\Z4d]e0de0fd^d_Z5ddPd`e0dae0ddPfdbdcZ6dudedfZ7deddfdgdhZ8d'dde#fdidjZ9dS)vN) dataclass) TYPE_CHECKINGCallableListOptionalTuple) is_in_test3RAY_DISABLE_CUSTOM_ARROW_JSON_OPTIONS_SERIALIZATION+RAY_DISABLE_CUSTOM_ARROW_DATA_SERIALIZATIONcCs>zddl}Wn tyYdSwt|t|t|dS)Nr)pyarrowModuleNotFoundError_register_arrow_data_serializer+_register_arrow_json_readoptions_serializer,_register_arrow_json_parseoptions_serializerserialization_contextparV/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/_private/arrow_serialization.py%_register_custom_datasets_serializerss   rcDtjtddkr dSddlm|jjddfddddS)N01rcSs |j|jfSN) use_threads block_sizeoptsrrr:s z=_register_arrow_json_readoptions_serializer..c j|Sr) ReadOptionsargspajsonrrr; custom_serializercustom_deserializer)osenvirongetr pyarrow.jsonjson _register_cloudpickle_serializerr rrr#rr,s   rcr)NrrrcSs|j|j|jfSr)explicit_schemanewlines_in_valuesunexpected_field_behaviorrrrrrMsz>_register_arrow_json_parseoptions_serializer..crr) ParseOptionsr!r#rrrRr%r&)r)r*r+r r,r-r.r3r/rr#rr?s   rcCs>tjtddkr dSddl}||jt||jt dS)ayCustom reducer for Arrow data that works around a zero-copy slicing pickling bug by using the Arrow IPC format for the underlying serialization. Background: Arrow has both array-level slicing and buffer-level slicing; both are zero-copy, but the former has a serialization bug where the entire buffer is serialized instead of just the slice, while the latter's serialization works as expected and only serializes the slice of the buffer. I.e., array-level slicing doesn't propagate the slice down to the buffer when serializing the array. We work around this by registering a custom cloudpickle reducers for Arrow Tables that delegates serialization to the Arrow IPC format; thankfully, Arrow's IPC serialization has fixed this buffer truncation bug. See https://issues.apache.org/jira/browse/ARROW-10739. rrNr) r)r*r+r r _register_cloudpickle_reducerTable_arrow_table_reduceSchema_arrow_schema_reducerrrrr Xs r schemazpyarrow.SchemareturnbytescCst|ffS)a;Custom reducer for Arrow Schema that uses IPC serialization for performance. Arrow's native IPC serialization for schemas is significantly faster than cloudpickle (10-20x for serialization, 2-3x for deserialization), making this optimization particularly valuable for workloads with large schemas. )_restore_schema_from_ipc serialize to_pybytesr9rrrr8rs r8bufcCsddl}|j||S)z7Restore an Arrow Schema serialized to Arrow IPC format.rN)r ipc read_schema BufferReader)r@rrrrr<sr<t pyarrow.Tablec Csg}|jD]S}||}zt|}WnAtyR}z5t|js$tr$|dt|jtvrBtjd|d|jdddt t|jt |WYd}~Sd}~ww| |qt ||j ffS)a/Custom reducer for Arrow Tables that works around a zero-copy slice pickling bug. Background: Arrow has both array-level slicing and buffer-level slicing; both are zero-copy, but the former has a serialization bug where the entire buffer is serialized instead of just the slice, while the latter's serialization works as expected and only serializes the slice of the buffer. I.e., array-level slicing doesn't propagate the slice down to the buffer when serializing the array. All that these copy methods do is, at serialization time, take the array-level slicing and translate them to buffer-level slicing, so only the buffer slice is sent over the wire instead of the entire buffer. See https://issues.apache.org/jira/browse/ARROW-10739. NzTFailed to complete optimized serialization of Arrow Table, serialization of column 'z ' of type z failed, so we're falling back to Arrow IPC serialization for the table. Note that this may result in slower serialization and more worker memory utilization. Serialization error:T)exc_info) column_names_arrow_chunked_array_reduce Exception_is_dense_uniontyper_serialization_fallback_setloggerwarningadd_arrow_table_ipc_reduceappend_reconstruct_tabler9)rDreduced_columns column_namecolumnreduced_columnerrrr6s.   r6rS pyarrow.Arraypyarrow.DataTypecCs:ddl}g}|D] \}}|t||q|jj||dS)zERestore a serialized Arrow Table, reconstructing each reduced column.rNr?)r rQ_reconstruct_chunked_arrayr5 from_arrays)rSr9rcolumnschunks_payloadtype_rrrrRs  rRcazpyarrow.ChunkedArrayPicklableArrayPayloadcCs.g}|jD] }t|}||q||jfS)zCustom reducer for Arrow ChunkedArrays that works around a zero-copy slice pickling bug. This reducer does not return a reconstruction function, since it's expected to be reconstructed by the Arrow Table reconstructor. )chunksr` from_arrayrQrK)r_chunk_payloadschunk chunk_payloadrrrrHs     rHrar^cCs"ddl}dd|D}|||S)z=Restore a serialized Arrow ChunkedArray from chunks and type.rNcSg|]}|qSrto_array).0rdrrr z._reconstruct_chunked_array..)r chunked_array)rar^rrrrrZs rZc@sbeZdZUdZded<eed<eded<eed<eed<eded <edd dZdddZ dS)r`zPicklable array payload, holding data buffers and array metadata. This is a helper container for pickling and reconstructing nested Arrow Arrays while ensuring that the buffers that underly zero-copy slice views are properly truncated. rYrKlengthpyarrow.Bufferbuffers null_countoffsetchildrenarXr:cCst|S)a Create a picklable array payload from an Arrow Array. This will recursively accumulate data buffer and metadata payloads that are ready for pickling; namely, the data buffers underlying zero-copy slice views will be properly truncated. )_array_to_array_payload)selfrsrrrrbsz PicklableArrayPayload.from_arraycCst|S)z7Reconstruct an Arrow Array from this picklable payload.)_array_payload_to_array)rurrrrhszPicklableArrayPayload.to_arrayNrsrXr:r`)r:rX) __name__ __module__ __qualname____doc____annotations__intr classmethodrbrhrrrrr`s     payloadc Csddl}dd|jD}|j|jr*t|dksJt||\}}|j||S|j|jrPt|dkrPt|dksCJt||\}}}|j |||St |j|j rmt|dkscJt||d}|j |S|j j|j|j|j|j|j|dS) zHReconstruct an Arrow Array from a possibly nested PicklableArrayPayload.rNcSrfrrg)ri child_payloadrrrrjrkz+_array_payload_to_array..rKrmrorprqrr)r rrtypes is_dictionaryrKlenDictionaryArrayr[is_mapMapArray isinstanceBaseExtensionType wrap_arrayArray from_buffersrmrorprq) rrrrindices dictionaryoffsetskeysitemsstoragerrrrvs,  rvrscCsddl}t|jr td|j|jrt|St|jr!t|St |jr*t |S|j |js8|j |jr.r) rorrprrqrangerK num_fieldsr`)rsrorrrrrrrs rpyarrow.UnionArraycsddl}tjr J}t|dksJt||d}|dus'J||d}t||jt}fddtjj D}t jt||gj d|dS)z0Serialize union arrays to PicklableArrayPayload.rNrcrrrrrrrrj'rz1_union_array_to_array_payload..r) r rJrKrorrint8rqrrr`rp)rsrror type_code_bufrrrrrrs"rpyarrow.DictionaryArraycCs4t|j}t|j}t|jt|g|jd||gdS)z5Serialize dictionary arrays to PicklableArrayPayload.rr)rtrrr`rKrrp)rsindices_payloaddictionary_payloadrrrr2s  rpyarrow.MapArrayc Cs4ddl}|}t|dksJt||jdkr%t|d|jt|}nd}|g}|d}t||j|jt|\}}}t||j j rS| |t |j ||g}n9|}t|dkscJt||j|t|d||g} |j||} |j||} t | t | t | g}t|jt|||jd|dS)z.Serialize map arrays to PicklableArrayPayload.rNrrr)r rorrprrqrrKrlib ListArrayrQrtrrrrint32rrr`) rsrror new_buffersrrrrrrrrrrrrDsB   rpyarrow.ExtensionArraycCs(t|j}t|jt|g|jd|gdS)Nrr)rtrr`rKrrp)rsstorage_payloadrrrr|s rrnrqrmcCsTddl}|dur|j|rt|||}|S|dur|jdnd}t||||}|S)Copy buffer, if needed.rNr)r rrr bit_width_copy_normal_buffer_if_needed)r@r^rqrmrtype_bytewidthrrrrs r byte_widthcCs2||}||}|dks||jkr|||}|S)rr)sizer)r@rrqrm byte_offset byte_lengthrrrrs  rcCsV|d}|d}t||d}|dks||jkr)|||}|dkr)t|||}|S)z)Copy bit-packed binary buffer, if needed.rr)_bytes_for_bitsrr_align_bit_offset)r@rqrm bit_offsetrrrrrrs  rarr_typec Csddl}ddlm}|j|s"|j|s"|j|s"|j|r'|}n| }t ||||d}|j ||dd|g}|d }|d |} |||}|j|ra|j|dd}|d}||| fS)zvCopy the provided offsets buffer, returning the copied buffer and the offset + length of the underlying data. rNrF)safe)r pyarrow.computecomputerrrris_large_unicodeint64rrrras_pysubtractis_int32castro) r@rrqrmrpac offset_typerrrrrrrs*           rncCs |dd@S)zpRound up n to the nearest multiple of 8. This is used to get the byte-padded number of bits for n bits. ir)rrrrrs rrrcCs>ddl}|}t|tj}||L}||tj}||S)z}Align the bit offset into the buffer with the front of the buffer by shifting the buffer and eliminating the offset. rN)r r>r} from_bytessys byteorderto_bytes py_buffer)r@rrrbytes_ bytes_as_intrrrrs  rtablecCsdddlm}ddlm}|}|||jd }||Wdn1s&wYt|ffS)a4Custom reducer for Arrow Table that works around a zero-copy slicing pickling bug by using the Arrow IPC format for the underlying serialization. This is currently used as a fallback for unsupported types (or unknown bugs) for the manual buffer truncation workaround, e.g. for dense unions. r)RecordBatchStreamWriter)BufferOutputStreamr?N) pyarrow.ipcr pyarrow.librr9 write_table_restore_table_from_ipcgetvalue)rrr output_streamwrrrrrPs   rPcCs@ddlm}|| }|WdS1swYdS)z6Restore an Arrow Table serialized to Arrow IPC format.r)RecordBatchStreamReaderN)rrread_all)r@rreaderrrrrs  $rcCsddl}|j|o|jdkS)z1Whether the provided Arrow type is a dense union.rNdense)r rrmoderrrrrJsrJ)rDrE)rr`r:rXrw)rsrr:r`)rsrr:r`)rsrr:r`)rsrr:r`)rsrr:r`)rsrr:r`)rsrr:r`)rrE):loggingr)r dataclassesrtypingrrrrrray._private.utilsrr r r getLoggerrxrMsetrLrrrr r;r8r<r6rRrHrZr`rvrtboolrrrrrrrrrrrrr}rrrrrrrPrrJrrrrs     .     # ")  "  ! 8      %