&`iOddZddlZddlmZmZddlmZmZmZm Z m Z m Z m Z ddl mZddlmZddlmZmZddlmZmZdd lmZdd lmZer"ddlZdd lmZdd lm Z dd l!m"Z"ddl#m$Z$ddl%m&Z&ej'e(Z)eGddZ*dZ+eGddee*Z,dS)zU Module to write a Ray Dataset into an iceberg table, by using the Ray Datasink API. N) dataclassfield) TYPE_CHECKINGAnyDictIterableListOptionalUnion) TaskContext)SaveMode)Block BlockAccessor)Datasink WriteResult)Expr) DeveloperAPI)Catalog)DataFile)Schema)Table) UpdateSchemaceZdZUdZeeZeded<dZ e e e ee fed<eeZeded<dS) IcebergWriteResultaIResult from writing blocks to Iceberg storage. Attributes: data_files: List of DataFile objects containing metadata about written Parquet files. upsert_keys: Dictionary mapping column names to lists of key values for upsert operations. schemas: List of PyArrow schemas from all non-empty blocks. )default_factoryr data_filesN upsert_keys pa.Schemaschemas)__name__ __module__ __qualname____doc__rlistrr __annotations__rr rstrrr/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/datasource/iceberg_datasink.pyrrs$)5#>#>#>JZ >>>26K$sDI~./666!&t!B$D$D$F$F! !1,288::"2"8b!>!>!@!@   4:#@#@maeakmm   ! djH4F&F&Fsgkgqss   ! djH4F&F&Fsgkgqss  &   !M6%))->>JP PPPP T) ) )!%!5!9!9&!A!AD  !*D # r(returncd|j}|dd|S)z!Exclude `_table` during pickling.rFN)__dict__r8rBrGstates r) __getstate__zIcebergDatasink.__getstate__s. ""$$ (D!!! r(rOcH|j|d|_dSN)rMupdaterFrNs r) __setstate__zIcebergDatasink.__setstate__s# U### r(rc>ddlm}|j|jfi|jS)Nr)catalog) pyicebergrV load_catalogrEr9rGrVs r) _get_catalogzIcebergDatasink._get_catalogs5%%%%%%#w#D$6OO$:NOOOr(cl|}||j|_dS)z*Reload the Iceberg table from the catalog.N)rZ load_tabler-rFrYs r) _reload_tablezIcebergDatasink._reload_tables/##%%(()>?? r(c&|jtg}|sn|jjjD]J}|jj|}|r||K|S)zGGet join columns for upsert, using table identifier fields as fallback.) r=get_UPSERT_COLS_IDrFmetadataschemaidentifier_field_idsfind_column_nameappend)rG upsert_colsfield_idcol_names r)_get_upsert_colsz IcebergDatasink._get_upsert_colss)--orBB  1 K07799N 1 1;/6688II(SS1&&x000r(txnzTable.transactionrrc||j5}|D]}|| dddn #1swxYwY|dS)zAppend data files to a transaction and commit. Args: txn: PyIceberg transaction object data_files: List of DataFile objects to append N)_append_snapshot_producerr:append_data_filecommit_transaction)rGrjr append_files data_files r)_append_and_commitz"IcebergDatasink._append_and_commits * *4+D E E 9' 9 9 --i8888 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9      sAA Aupsert_keys_dictscddl}ddlm}ddlddlm}|t }|D]7}|D] \} } || | !8|rj t|| } fd| D} | j j| } | t!dkrU|| }|j}|t(d|jd||jd||||dS)a  Commit upsert transaction with copy-on-write strategy. Args: txn: PyIceberg transaction object data_files: List of DataFile objects to commit upsert_keys_dicts: List of dictionaries mapping column names to lists of key values rN defaultdict)create_match_filterc3XK|]$}j|V%dSrR)computeis_valid).0col keys_tablepas r) z1IcebergDatasink._commit_upsert..s7QQcRZ((C99QQQQQQr(r5r/r') functools collectionsrupyarrowpyiceberg.table.upsert_utilrvr$itemsextendtabledictrireducerxand_filterlenr=r8rBr`deleter:rq)rGrjrrrrrurvmerged_keys_dictupsert_keys_dictr{valuesrfmasksmaskr5 delete_kwargsr|r}s @@r)_commit_upsertzIcebergDatasink._commit_upserts ++++++CCCCCC';t,, 1 5 5 /5577 5 5 V %,,V4444 5  !$'7"8"899J//11KQQQQQ[QQQE##BJOU;;D#**400J:"" 3 3J L L !% 3 8 8 : : !!/4888 "/(,(A$ Z00000r(rSr table_schemarc ddlm}|j}|D]W}||jvrL|j|}|js8||j|j|j|jd|j |j |j|<XdS)aEnsure identifier fields remain required after schema union. When union_by_name is called with a schema that has nullable fields, PyIceberg may make identifier fields optional. Since identifier fields must be required, this helper ensures they remain required after union. Example: Table schema: id: int (required, identifier), val: string Input schema: id: int (optional), val: string `union_by_name` merges them to: id: int (optional), val: string This violates the identifier constraint. This function forces `id` back to required in the pending update. Args: update: The UpdateSchema object from update_schema() context manager table_schema: The current table schema to get identifier field IDs from r) NestedFieldT)rgr6 field_typedocrequiredinitial_default write_defaultN) pyiceberg.typesrrc_updatesrrgr6rrrr)rGrSrrrcrg updated_fields r)'_preserve_identifier_field_requirementsz7IcebergDatasink._preserve_identifier_field_requirementss. 0/////+@,  H6?** & 9 $- 1< !.!7*/#0#;)-!%(5(E&3&A111FOH-  r( new_schema)rrc\|||||dS)aUpdate schema using union_by_name while preserving identifier field requirements. Args: update: The UpdateSchema object. new_schema: The new schema to union with the table schema. table_schema: The current table schema. N) union_by_namer)rGrSrrs r)_update_schema_with_unionz)IcebergDatasink._update_schema_with_unions4 Z((( 44V\JJJJJr(rbrc||z|jj}|j5}||||dddn #1swxYwY||jtjkrV|j tg}|s6|jjj }|stddSdSdS)apInitialize table for writing and create a shared write UUID. Args: schema: The PyArrow schema of the data being written. This is automatically extracted from the first input bundle by the Write operator. Used to evolve the table schema before writing to avoid PyIceberg name mapping errors. Nz`join_cols must be specified in upsert_kwargs for UPSERT mode when table has no identifier fields)r]rFrarb update_schemarr;r r?r=r_r`rcr@)rGrbrrSrfrcs r)on_write_startzIcebergDatasink.on_write_start/sb   ;/6688L**,, M..vv|LLL M M M M M M M M M M M M M M M     : ( (-11/2FFK K(//11F%,$> ) (   sA22A69A6blocksctxcddlm}ddlm}|j|g}|t }g}|jtj k}|D]} tj |  } | j dkr|| j|rL|} | D]5} || | | 6t ||jj| |jj} || t+||pd|S)aD Write blocks to Parquet files in storage and return DataFile metadata with schemas. This runs on each worker in parallel. Files are written directly to storage (S3, HDFS, etc.) and only metadata is returned to the driver. Schema updates are NOT performed here - they happen on the driver. Args: blocks: Iterable of Ray Data blocks to write ctx: TaskContext object containing task-specific information Returns: IcebergWriteResult containing DataFile objects, upsert keys, and schemas. rrt)_dataframe_to_data_filesN)table_metadatadfio)rrr)rrupyiceberg.io.pyarrowrrFr]r$r;r r?r for_blockto_arrownum_rowsrerbrir to_pylistrarr)rGrrrurall_data_filesr block_schemasuse_copy_on_write_upsertblockpa_tablerfr{rs r)writezIcebergDatasink.writeQs ,+++++AAAAAA ;     &;t,, #':#@  2 2E$.u55>>@@H 1$$$$X_555,P"&"7"7"9"9K*PP(-44Xc]5L5L5N5NOOOO",,'+{';#;> %%j111!%(0D!    r(c|jEddlm}|}||j}|jd||jd|jn(ddlm}|jd||jd|j| ||dS)z'Commit data files using OVERWRITE mode.Nr)_IcebergExpressionVisitorr) AlwaysTruer') r<0ray.data._internal.datasource.iceberg_datasourcervisitrr:r>pyiceberg.expressionsrrq)rGrjrrvisitor pyi_filterrs r)_commit_overwritez!IcebergDatasink._commit_overwrites  ! -      0/11G t'=>>J CJ ($($=  (     9 8 8 8 8 8 CJ (jll$($=  (    Z00000r( write_resultcg}g}g}|jD]a}|s|jrU||j||j|jr||jb|sdSddlm}ddlm }| |j }||g|zd} |j } | |sa|j j } | 5} || | | dddn #1swxYwY|jt&jkr|| |dS|jt&jkr|| |dS|jt&jkr|| ||dSt5d|j)ak Complete the write by reconciling schemas and committing all data files. This runs on the driver after all workers finish writing files. Collects all DataFile objects and schemas from all workers, reconciles schemas (allowing type promotion), updates table schema if needed, then performs a single atomic commit. Nr)r) unify_schemasT) promote_typeszUnsupported mode: ) write_returnsrrrrre pyiceberg.ior.ray.data._internal.arrow_ops.transform_pyarrowrschema_to_pyarrowrFrb transactionequalsrarrr;r APPENDrqrArr?rr@) rGrr all_schemasrr write_return pyi_pa_iorrfinal_reconciled_schemarjcurrent_table_schemarSs r)on_write_completez!IcebergDatasink.on_write_completesq,.)+ 8:(6 G GL & G%%l&=>>>""<#7888+G%,,\-EFFF  F 655555PPPPPP 224;3E3E3G3GHH "/- N[ (# # #  k%%'''--l;; #';#7#>#>#@#@ ""$$ ..35I                : ( (  # #C 8 8 8 8 8 Z8- - -  " "3 7 7 7 7 7 Z8? * *   ^5F G G G G G>$*>>?? ?sEEE)rKr)rKN)rSrrrrKNrR) r r!r"r#r rr&r rrrJrrPrTrZr]r rirqrrr rrrrr rrrrrr'r(r)r,r,,s  488<!-12659P$P$P$!c3h0P$&d38n5 P$  P$ #6* P$ S#X/P$#4S>2P$P$P$P$dd $4PPPP @@@@ $s)     !& !484D !  ! ! ! !61 61$61 S$s)^ 45 61  61616161p-$-4<- ----^KK/0K K  KKKK  Xk%: d    D6 HUO6 +6 BT6 6 6 6 p1&1484D1 1111>9@k9@d9@9@9@9@9@9@r(r,)-r#logging dataclassesrrtypingrrrrr r r 'ray.data._internal.execution.interfacesr ray.data._internal.savemoder ray.data.blockrrray.data.datasource.datasinkrrray.data.expressionsrray.util.annotationsrrr}pyiceberg.catalogrpyiceberg.manifestrpyiceberg.schemarpyiceberg.tablerpyiceberg.table.update.schemar getLoggerr rCrr`r,r'r(r)rs((((((((LLLLLLLLLLLLLLLLLL??????000000////////>>>>>>>>%%%%%%------;))))))++++++''''''%%%%%%::::::  8 $ $  = = = = = = =  =t@t@t@t@t@h12t@t@t@t@t@r(