o (sz2create_match_filter...) functoolsreduceoperatorand_)rr rrr'sz'create_match_filter..)selectgroup_by aggregatelenr to_pylistrr)r r unique_keysfiltersr rrcreate_match_filter!s    r cCs4t|||gdfgtddkdkS)zFCheck for duplicate rows in a PyArrow table based on the join columns. count_allr r)rrrrfilterpcfield)r r r r rhas_duplicate_rows3s4r% source_table target_tablecCspt|j}t|}t||}t||rtdt|dkr#|jSd}d}||vs/||vr:t|d|dd||j | |t t t|}| | |t t t|} |j| t|dd } g} t| || |d d D]2\} } || d }|| d }|D]}||d}||d}||kr| | nqqy| r|| S|jS) a  Return a table with rows that need to be updated in the target table based on the join columns. The table is joined on the identifier columns, and then checked if there are any updated rows. Those are selected and everything is renamed correctly. z0Target table has duplicate rows, aborting upsertr__source_index__target_indexz and zH are reserved for joining DataFrames, and cannot be used as column namesNinner)keys join_typeT)strictr )set column_nameslistr% ValueErrorrschema empty_tablecastr append_columnpaarrayrangejoinziprslicecolumnas_pyappendtake)r&r'r all_columns join_cols_set non_key_colsSOURCE_INDEX_COLUMN_NAMETARGET_INDEX_COLUMN_NAME source_index target_indexmatching_indicesto_update_indices source_idx target_idx source_row target_rowkey source_val target_valr r rget_rows_to_update8sP              rP)rrpyarrowr6r pyarrow_tablerr#pyiceberg.expressionsrrrrrr0strr boolr%rPr r r rs   (