o $iKl @sddlZddlZddlZddlmZddlmZmZmZm Z ddl m Z er*ddl Z dZ dZeeZeaddZd d Zd d Zd dZdfddZdee eddfddddfddZddde eddffddZd edd!dddfd"d#ZeGd$ddZdgd&d'Zdhd)d*Zd!ddefd+d,Z d!ddefd-d.Z!did0d1Z"dhd2d3Z#dhd4d5Z$dhd6d7Z% ( 8  djd9d:Z&dkdzddl}Wn tyYdSwt|t|t|dS)Nr)pyarrowModuleNotFoundError_register_arrow_data_serializer+_register_arrow_json_readoptions_serializer,_register_arrow_json_parseoptions_serializerserialization_contextpar]/home/ubuntu/veenaModal/venv/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_serializerrrrr"rr ,s   r cr)NrrrcSs|j|j|jfSr)explicit_schemanewlines_in_valuesunexpected_field_behaviorrrrrrMsz>_register_arrow_json_parseoptions_serializer..crr) ParseOptionsr r"rrrRr$r%)r(r)r*rr+r,r-r2r.rr"rr?s   rcCs0tjtddkr dSddl}||jtdS)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_reducerrrrr Xsr 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_tableschema)r6reduced_columns column_namecolumnreduced_columnerrrr5qs.   r5rF pyarrow.Arraypyarrow.DataTyperEzpyarrow.SchemareturncCs:ddl}g}|D] \}}|t||q|jj||dS)zERestore a serialized Arrow Table, reconstructing each reduced column.rNrE)r rC_reconstruct_chunked_arrayr4 from_arrays)rFrErcolumnschunks_payloadtype_rrrrDs  rDcazpyarrow.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. )chunksrU from_arrayrCr=)rTchunk_payloadschunk chunk_payloadrrrr:s     r:rVrScCs"ddl}dd|D}|||S)z=Restore a serialized Arrow ChunkedArray from chunks and type.rNcSg|]}|qSrto_array).0rYrrr z._reconstruct_chunked_array..)r chunked_array)rVrSrrrrrOs rOc@sbeZdZUdZded<eed<eded<eed<eed<eded <edd dZdddZ dS)rUzPicklable 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. rLr=lengthpyarrow.Bufferbuffers null_countoffsetchildrenarKrMcCst|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)selfrhrrrrWsz PicklableArrayPayload.from_arraycCst|S)z7Reconstruct an Arrow Array from this picklable payload.)_array_payload_to_array)rjrrrr]szPicklableArrayPayload.to_arrayNrhrKrMrU)rMrK) __name__ __module__ __qualname____doc____annotations__intr classmethodrWr]rrrrrUs     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.rNcSr[rr\)r^ child_payloadrrrr_r`z+_array_payload_to_array..r=rbrdrerfrg)r rgtypes is_dictionaryr=lenDictionaryArrayrPis_mapMapArray isinstanceBaseExtensionType wrap_arrayArray from_buffersrbrdrerf) rtrrgindices dictionaryoffsetskeysitemsstoragerrrrks,  rkrhcCsddl}t|jr td|j|jrt|St|jr!t|St |jr*t |S|j |js8|j |jr.ry) rdr|rerrfranger= num_fieldsrU)rhrdrrgrrrrs rpyarrow.UnionArraycsddl}tjr J}t|dksJt||d}|dus'J||d}t||jt}fddtjj D}t jt||gj d|dS)z0Serialize union arrays to PicklableArrayPayload.rNrwcrrrrrrrr_rz1_union_array_to_array_payload..ry) r r<r=rdr|rint8rfrrrUre)rhrrdr type_code_bufrgrrrrs"rpyarrow.DictionaryArraycCs4t|j}t|j}t|jt|g|jd||gdS)z5Serialize dictionary arrays to PicklableArrayPayload.rry)rirrrUr=r|re)rhindices_payloaddictionary_payloadrrrrs  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.rNrwrvry)r rdr|rerrfrr=rlib ListArrayrCrirrrrint32rrrU) rhrrdr new_buffersrrrrgrrrrrrr/sB   rpyarrow.ExtensionArraycCs(t|j}t|jt|g|jd|gdS)Nrry)rirrUr=r|re)rhstorage_payloadrrrrgs rbufrcrfrbcCsTddl}|dur|j|rt|||}|S|dur|jdnd}t||||}|S)Copy buffer, if needed.rNrw)r rzrr bit_width_copy_normal_buffer_if_needed)rrSrfrbrtype_bytewidthrrrrus r byte_widthcCs2||}||}|dks||jkr|||}|S)rr)sizer)rrrfrb 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)rrfrb 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. rNrwF)safe)r pyarrow.computecomputerzrrris_large_unicodeint64rrrras_pysubtractis_int32castrd) rrrfrbrpac 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 to_pybytesrr from_bytessys byteorderto_bytes py_buffer)rrrrbytes_ 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)BufferOutputStreamrNN) pyarrow.ipcr pyarrow.librrE write_table_restore_table_from_ipcgetvalue)rrr output_streamwrrrrrBs   rBcCs@ddlm}|| }|WdS1swYdS)z6Restore an Arrow Table serialized to Arrow IPC format.r)RecordBatchStreamReaderN)rrread_all)rrreaderrrrrs  $rcCsddl}|j|o|jdkS)z1Whether the provided Arrow type is a dense union.rNdense)r rzrmoderrrrr<sr<)r6r7)rtrUrMrKrl)rhrrMrU)rhrrMrU)rhrrMrU)rhrrMrU)rhrrMrU)rhrrMrU)rhrrMrU)rr7)7loggingr(r dataclassesrtypingrrrrray._private.utilsrr rr getLoggerrmr?setr>rr rr r5rDr:rOrUrkriboolrrrrrrrrrrrrrrrrrrrrrBbytesrr<rrrrs    .     # ")  "  ! 8      %