bi RddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z ddlmZmZddlZddlZddlZddlmZddlZddlmZddlmZddl m!Z"ddl#Z$ddl%m&Z&ddl$m'Z'ddl(Z(ddl)m*Z*m+Z+dd l,m-Z-dd l.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>dd l?m@Z@dd lAmBZBdd lCmDZDmEZEddlFmGZGddlHmIZImJZJmKZKmLZLmMZMmNZNmOZOddlPmQZQddlRmSZSddlTmUZUddlVmWZWmXZXmYZYddlZm[Z[ddl\m]Z]m^Z^m_Z_m`Z`maZaeXjej<jdZddaedZfiZgejejejejejejejejejejejejejejejejejejejejejejejeji ZdZdZiZGddZdZeXjej dZeXjejdZeXjejd ZeXje"jd!ZeXje"jd"ZGd#d$e7ZGd%d&e6ZGd'd(e7Z dCd)Zd*ZGd+d,eEeDZGd-d.eZGd/d0eZdDd1Zd2Zd3Zd4ZGd5d6Z dE dFd7ZdEd8Zd9Zd:Zd;Zd<Zd=Zd>Zd?Ze(jd@ZdAZdBZy)G) annotationsN)abstractmethod) defaultdict)cached_propertypartial)stringify_path)identity)TaskTaskRef)flatten)EQGEGTLELTNEAnd BlockwiseExprFilterIndexLengthsLiteralOr Projectiondetermine_column_projection)Len)_convert_to_list) BlockwiseIOPartitionsFiltered)FusedParquetIO)ParquetFunctionWrapperToParquetFunctionWrapperaggregate_row_groups apply_filters get_engineset_index_columnssorted_columns)_split_user_options) _is_local_fs)delayed)_tokenize_deterministicnormalize_tokentokenize)Key)applyfuncnamenatural_sort_key parse_bytestypenamecrt|j|j|j|jffSN)type__name__pathsizemtime_ns)fileinfos ^/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/io/parquet.py_tokenize_fileinfor>Es5 > " "  % Fcdts*tjtjdayy)NT)_CPU_COUNT_SETpa set_cpu_countos cpu_countr?r=_maybe_adjust_cpu_countrGQs%  ( r?__null_dask_index__ cheZdZejZdZ d d dZdZdZ e dZ dZ dZ y) FragmentWrapper)NNcJ||_||_||_d|_||_y)z8Wrap a pyarrow Fragment to only deserialize when needed.N) _fragment_fragment_packed _file_size_fs _filesystem)selffragment filesystem file_sizefragment_packeds r=__init__zFragmentWrapper.__init__ws) " /#%r?c@|j|jj}|jt j drd}|jj }|jjtjjrd}|jxs|jj}|j|jjsJ|jdt|k7r)t|tj|ft _ t jd}||jj"|jj$n|jj"|||j&f|_dx|_|_y)NTr)rOrNpartition_expressionequalspcscalarformat read_optionspa_dsParquetFileFormatrRrU_filesystem_pickle_cacheidpickledumpsrKbufferr9rPrQ)rS part_exprpqformatfsfs_pkls r=packzFragmentWrapper.packs?  (;;I $0 ~~,,H$$++E,C,C,E,R,RS!!>T^^%>%>B99T^^667 77,,Q/2b69<>rFFLLQSDT;U8$==a@F~~,,4NN''.. %D !%)(4>r?c|j|j\}}}}}tjj |}|(t j |}|tj|<|durtjd}||_ |tj}|j|||||_d|_y)NT)rUr[rV) rNrOrK _filesystemsgetreloadsr]r^rQrarb make_fragment)rSripath_or_bufferfs_rawr[rVrjs r=unpackzFragmentWrapper.unpacks >> !%%  $ --11&9Bz\\&)79,,V4#t+')yy$DH 224%33%9# 4DN !%r?c:|j|jSr6)rtrNrSs r=rTzFragmentWrapper.fragments ~~r?ct|jt|j|j|j ffSr6)r7r8r-rTrOrPrvs r=__dask_tokenize__z!FragmentWrapper.__dask_tokenize__s<Dz""O %% %   r?cN|jtddd|jffSr6)rlrKrOrvs r= __reduce__zFragmentWrapper.__reduce__s% tT43H3H IIIr?NN)returnNone)r8 __module__ __qualname__weakrefWeakValueDictionaryrnrcrXrlrtpropertyrTrxrzrFr?r=rKrKssU.7..0L)EI &  &)6%8 Jr?rKctttkDr?|tvr6ttj d}tj |yyyNr)len _cached_plan_CACHED_PLAN_SIZElistkeyspop)key key_to_pops r=_control_cached_planrsF <,,L1H,++-.q1 $2I,r?c2|j|jfSr6)filesschema)dss r=normalize_pa_dsrs HHbii  r?ct|Sr6)str) file_formats r=normalize_pa_file_formatrs { r?c"|jSr6) to_stringrs r=normalize_pa_schemars    r?c` t|S#t$rtt|cYSwxYwr6)hash TypeErrorreprrs r=normalize_pq_schemars/"F| "DL!!"s --c t|S#t$rNtt|j|j|j |j |jfcYSwxYwr6)rrrrnum_rowsnum_row_groupsformat_versionserialized_size)metas r=normalize_pq_filemetadatarsb Dz   T[[! ####$$     s AA$#A$c0eZdZgdZedZdZdZy) ToParquet) framer9rjfmdengineoffset partition_onwrite_metadata_file name_function write_kwargsappendcyr6rFrvs r=_metazToParquet._meta r?cyNr{rFrvs r= _divisionszToParquet._divisionsr?cTtt|jg|jddSNrZ)ToParquetBarrier ToParquetDataoperandsrvs r=_lowerzToParquet._lowers3   ]]12    r?N)r8r~r _parametersrrrrrFr?r=rrs' K r?rcBeZdZejZedZdZddZy)rc t|j|j|j|j|j |j |j|jSr6) r#rr9rjrrrrrrvs r=io_funczToParquetData.io_funcsO' KK II GG     $ $ KK        r?c:d|jjdzzS)Nr6rZ)r npartitionsrvs r=rzToParquetData._divisions+s$**001455r?crt||jt|jj|f|fSr6)r rr r_name)rSnameindexs r=_taskzToParquetData._task.s-D$,,1A1A50I(JUHUUr?Nrr/rintr|r ) r8r~rrrrrrrrFr?r=rrs*''K     6Vr?rc@eZdZejZedZdZdZy)rcyr6rFrvs r=rzToParquetBarrier._meta5rr?cyrrFrvs r=rzToParquetBarrier._divisions9rr?c|jr|j}|jjd}|jdft |j j|jj|j|j|jg||dfiS|jdfd|jjfiS)N compressionr)rrcyr6rFxs r=z)ToParquetBarrier._layer..Nsr?) rrrrorr0rwrite_metadatar __dask_keys__rrjr9)rSrrs r=_layerzToParquetBarrier._layer<s  # #[[F++// >KQKK.. 002  &kB "  ZZOndjj6N6N6P%QR Rr?N) r8r~rrrrrrrrFr?r=rr2s*''K Sr?rc  ddlm}t||j}| xsi} |xsg}t |t r|g}t |t |jz r6tdt |dt t|j|jjdvr tdt |t r t|}t|dr t|}|j||ii| \}}}}t|d k(sJd |d}|r|r td |j!|r|j#|r|j$j't(D]Y}t |j*j-d d z}|j-d d z}|j/|sPtdt1|rB|j3dd}|j-d |j-d k(r td|j5|dt6j9|j:sd}|j<|j>j@d}|d>tB|jvr tB|d<n"|r tdtEjFdg}dD]"}||jvs|jI|$g}|rt |j}t|jj>jJdgk(}|jM}|r5|jD cic]} | |vs| tB}!} |jO|!}t |j|z D cgc]} | }} n|jMd}| rd| jQvr td|jR|||f|||||| | d|\}"}#}$}%|r| |$} |ltU|s tdtW|jXD&cgc] }&||&|"z}'}&tt |'t|'kr tdtZj\j_di}(d |(vr"t1|st[j`d!"})ntcjd})tEjf5tEjhd#d$tj%|)5|tm||||#||"|| |tojp||| d&|%| }*dddddd| r*jrd'i| }*|ju|*Scc} wcc} wcc}&w#1swYGxYw#1swYKxYw)(u9Store Dask.dataframe to Parquet files Notes ----- Each partition will be written to a separate file. Parameters ---------- df : dask.dataframe.DataFrame path : string or pathlib.Path Destination directory for data. Prepend with protocol like ``s3://`` or ``hdfs://`` for remote data. compression : string or dict, default 'snappy' Either a string like ``"snappy"`` or a dictionary mapping column names to compressors like ``{"name": "gzip", "values": "snappy"}``. Defaults to ``"snappy"``. write_index : bool, default True Whether or not to write the index. Defaults to True. append : bool, default False If False (default), construct data-set from scratch. If True, add new row-group(s) to an existing data-set. In the latter case, the data-set must exist, and the schema must match the input data. overwrite : bool, default False Whether or not to remove the contents of `path` before writing the dataset. The default is False. If True, the specified path must correspond to a directory (but not the current working directory). This option cannot be set to True if `append=True`. NOTE: `overwrite=True` will remove the original data even if the current write operation fails. Use at your own risk. ignore_divisions : bool, default False If False (default) raises error when previous divisions overlap with the new appended divisions. Ignored if append=False. partition_on : list, default None Construct directory-based partitioning by splitting on these fields' values. Each dask partition will result in one or more datafiles, there will be no global groupby. storage_options : dict, default None Key/value pairs to be passed on to the file-system backend, if any. custom_metadata : dict, default None Custom key/value metadata to include in all footer metadata (and in the global "_metadata" file, if applicable). Note that the custom metadata may not contain the reserved b"pandas" key. write_metadata_file : bool or None, default None Whether to write the special ``_metadata`` file. If ``None`` (the default), a ``_metadata`` file will only be written if ``append=True`` and the dataset already has a ``_metadata`` file. compute : bool, default True If ``True`` (default) then the result is computed immediately. If ``False`` then a ``dask.dataframe.Scalar`` object is returned for future computation. compute_kwargs : dict, default True Options to be passed in to the compute method schema : pyarrow.Schema, dict, "infer", or None, default "infer" Global schema to use for the output dataset. Defaults to "infer", which will infer the schema from the dask dataframe metadata. This is usually sufficient for common schemas, but notably will fail for ``object`` dtype columns that contain things other than strings. These columns will require an explicit schema be specified. The schema for a subset of columns can be overridden by passing in a dict of column names to pyarrow types (for example ``schema={"field": pa.string()}``); columns not present in this dict will still be automatically inferred. Alternatively, a full ``pyarrow.Schema`` may be passed, in which case no schema inference will be done. Passing in ``schema=None`` will disable the use of a global file schema - each written file may use a different schema dependent on the dtypes of the corresponding partition. name_function : callable, default None Function to generate the filename for each output partition. The function should accept an integer (partition index) as input and return a string which will be used as the filename for the corresponding partition. Should preserve the lexicographic order of partitions. If not specified, files will created using the convention ``part.0.parquet``, ``part.1.parquet``, ``part.2.parquet``, ... and so on for each partition in the DataFrame. filesystem: "fsspec", "arrow", or fsspec.AbstractFileSystem backend to use. **kwargs : Extra options to be passed on to the specific backend. Examples -------- >>> df = dd.read_csv(...) # doctest: +SKIP >>> df.to_parquet('/path/to/output/', ...) # doctest: +SKIP By default, files will be created in the specified output directory using the convention ``part.0.parquet``, ``part.1.parquet``, ``part.2.parquet``, ... and so on for each partition in the DataFrame. To customize the names of each file, you can use the ``name_function=`` keyword argument. The function passed to ``name_function`` will be used to generate the filename for each partition and should expect a partition's index integer as input and return a string which will be used as the filename for the corresponding partition. Strings produced by ``name_function`` must preserve the order of their respective partition indices. For example: >>> name_function = lambda x: f"data-{x}.parquet" >>> df.to_parquet('/path/to/output/', name_function=name_function) # doctest: +SKIP will result in the following files being created:: /path/to/output/ ├── data-0.parquet ├── data-1.parquet ├── data-2.parquet └── ... See Also -------- read_parquet: Read parquet data to dask.dataframe r)new_collection)rrz2Partitioning on non-existent column. partition_on=z .columns=>emptystringz/parquet doesn't support non-string column namesr)rUdataset_optionsopen_file_optionsstorage_optionsrZz only one pathz3Cannot use both `overwrite=True` and `append=True`!/zICannot overwrite a path that you are reading from in the same task graph..z;Cannot clear the contents of the current working directory!T recursive) divisionsrNz:Index must have a name if __null_dask_index__ is a column.zfIf read back by Dask, column named __null_dask_index__ will be set to the index (and renamed to None).)rlevel_0columns)dropspandaszUser-defined key/value metadata (custom_metadata) can not contain a b'pandas' key. This key is reserved by Pandas, and overwriting the corresponding value can render the entire dataset unreadable.)rignore_divisionsr division_info index_colsrcustom_metadataz7``name_function`` must be a callable with one argument.z0``name_function`` must produce unique filenames.rretries)rignorezDask annotations )messagecategory)rrrF);$dask.dataframe.dask_expr._collectionr_set_parquet_enginer isinstancersetr ValueErrorr inferred_typer&hasattrrextract_filesystemrexistsisdirexprfind_operations ReadParquetr9rstrip startswithr* expand_pathrmrclearknown_divisionsrrr NONE_LABELwarningswarnrnames reset_indexrenamerinitialize_writecallablerangerdaskconfigroannotate contextlib nullcontextcatch_warningsfilterwarnings UserWarningrtoolzmergecomputeinvalidate_cache)+dfr9r write_indexr overwriterrrrrrcompute_kwargsrrrUrkwargsrrj_paths_read_opread_path_with_slashwrite_path_with_slash working_dirrreserved_namesrr real_cols none_indexcrename_columnsi_offsetrmetadata_file_existsextra_write_kwargsi filenamesrctxouts+ r= to_parquetr0QspBD RXX >F#)rN%2L,$$~  <3rzz?* -s4 3C/D F   zz'::JKK&#F#tVd#00 ' 1B1 v;! ,_,  !9D RS S 99T?rxx~7722;? '*7<<'8'?'?'Ds'J$(, C(83(>%'223HI$7  B nnS1!4 ;;s#{'9'9#'>>$U EE$$E '    #%,, FMV$ RZZ '$.M& ! L  MMB N$( rzz !  ! !$ '(J O "((..../D69 ^^  57ZZW1CVamWNW>2B!$RZZ9!<=Aa= =^^^ &9(<(<(>> )  ?Vf>U>U    ?)!#' ?  ?;Hc'); %-2  &VW W:?:OPQ]1x<0P P s9~ Y /OP P ++//-4K #L,<mmA&$$&  " 1K    '!KK+6/>+ %C  8ckk+N+  JwX=NQ$   s< U U UUU)#4UU)U& "U))U2cgd}|j||rjtjtjdij jtj tjtj ij jtjtjtjij d}j||dk(r jtj n|dk(rj|fd}tdkDr|Sy)Nc|tjk(rtjdStj|S)Npyarrow)rBrpd StringDtype ArrowDtype) pyarrow_dtypes r=pyarrow_type_mapperz3_determine_type_mapper..pyarrow_type_mappers0 BIIK '>>), ,==/ /r?r3cltjj|rtj|Syr6)rBtypes is_decimalr4r6)r7s r=_convert_decimal_typez5_determine_type_mapper.._convert_decimal_types&xx""4(}}T**r?numpy_nullablec0D]}||}||cSy)zBTry all type mappers in order, starting from the user type mapper.NrF)r7type_converterconverted_type type_mapperss r=default_types_mapperz4_determine_type_mapper..default_types_mappers(* &N+M:N)%% &r?r) rrBrr4r5rodate32r6date64PYARROW_NULLABLE_DTYPE_MAPPINGr)user_types_mapper dtype_backendpyarrow_strings_enabledr8r<rBrAs @r=_determine_type_mapperrIsL0$-.RYY["..*CDHHIRYY["-- *DEIIJRYY["-- *DEIIJ  12((:>>? ) #/0& <1##r?ceZdZdZdZdZfdZfdZedZ e dZ dZ e dZ ed Zfd Ze d Zed Zed ZxZS)rFTc t|||xrlt|jtt t tttttfxr,tj||jjduSr6)super_filter_passthrough_availabler predicaterrrrr rrr_DNFextract_pq_filters_filters)rSparent dependents __class__s r=rMz)ReadParquet._filter_passthrough_availablesf G 1&* E UF,,r2r2r2sB.OP U''f.>.>?HHPTT r?c&t|trjt|||}|jDcgc] }||vs| }}t |t |jk(ryt|j |ddSt|t rt|!||St|tr}|j||rktj||j}|j?|j d|j|j!dj#iSt|t$r|j'}|r t)|St|t*r'|j'}|rt)t-|Syycc}w)NF)r_seriesfilters)rrrrrsubstitute_parametersrrL _simplify_uprrMrOrPrNrQcombineoperand to_list_tupler _get_lengthsrrsum)rSrRrSrcolrW_lengthsrTs r=rYzReadParquet._simplify_upsk fe $1$ KG&*llEscWnsEGE7|s4<<00**w5+QR  fj )7' ; ; ff %$*L*L J+ --dF4D4DEG+11!7?? LL3$'-/ fg &((*Hx(( fc "((*Hs8}-- #;Fs FFc||jd}|t|jjSt |S)Nr)r[rrrr)rScolumns_operands r=rzReadParquet.columnss6,,y1  " **+ +#O4 4r?cy)N read_parquetrFrvs r= _funcnamezReadParquet._funcnamesr?c|js;ttt||jg|j dd|_|jS)NrL) _determ_tokenr,r1r7checksumrrvs r=rxzReadParquet.__dask_tokenize__ sK!!!8d$dmm"6:mmCR6H"D !!!r?c:|jdz|jzS)N-)redeterministic_tokenrvs r=rzReadParquet._names~~#d&>&>>>r?c |jdS)Nrh _dataset_inforvs r=rhzReadParquet.checksums!!*--r?cN|j|dk(r|St| |||S)N_dataset_info_cache)rrL _tree_repr_argument_construction)rSr,opheaderrTs r=rqz,ReadParquet._tree_repr_argument_constructions0   A "7 7Mw72vFFr?c|jd}t|jd}|jrt |dkDsJ||dS|||S|S)N base_metarr)rnrr[rVr)rSrrs r=rzReadParquet._meta"sb!!+."4<< #:; <<w.s !&&r?rZ)ninnoutr) rrrrn frompyfuncargsortrrzr rr)rSnfragsfinfosgetsizefinfo_size_arr finfo_argsortnfragsstepsizefinfos_sampled frags_samplesixsr,sort_ixs r=rz&ReadParquetPyarrowFS.sample_statisticss''$,,[9:-- 0aa@ &..0 Uv{A& q&(+ 1A#A&G!(6!1gqG JJw   ! !&/ 2  w 0  1 ^]; r?c|j|jdDcgc]}tt|c}Scc}w)zParquet statstics for every file in the dataset. The statistics do not include all the metadata that is stored in the file but only a subset. See also `_extract_stats`. r)rrnrr.)rSrs r=raw_statisticsz#ReadParquetPyarrowFS.raw_statisticssA 7;7I7I+7V .3L% )   s>c,t|jS)aAggregate statistics for every partition in the dataset. These statistics aggregated the row group statistics to partition level such that min/max/total_compressed_size/etc. corresponds to the entire partition instead of individual row groups. )_aggregate_statistics_to_filerrvs r=aggregated_statisticsz*ReadParquetPyarrowFS.aggregated_statisticss-T-@-@AAr?cT|jstd|jDSy)Nc3&K|] }|d yw)rNrF).0rs r= z4ReadParquetPyarrowFS._get_lengths..sSuz*S)rWtuplerrvs r=r]z!ReadParquetPyarrowFS._get_lengthss%||S8R8RSS Sr?c|jdx}r|Si}|j} g}|D]}tj|d}|j |j j |Dcgc]5}|jtj jjk(r|7c} d}d} d} |jst|d}|djj!dr|j#}t%|} t'j(|j*|j } d|d <| j-D cgc]} t/| |j c} |d <dgt1|d z|d <| (t%|} |D cgc]} | j2c} |d <| |d <| ddlm} | j8|D cgc]} | j*c} |j |j:} d|d <| j<D cgc]} t/| |j c} |d <||d<| j>|d<| j>jAjC|d<||_"|Scc}w#ttf$r2|Dcgc]}|j j |ncc}w}}YwxYwcc} wcc} wcc} wcc} w)NrpTrFc8|jjdS)N _metadata) base_nameendswithrs r=rz4ReadParquetPyarrowFS._dataset_info.. s)=)=k)Jr?rrLrrUusing_metadata_filer file_sizesrhr)rUrWrrru)#r[rr FileSelectorextendrj get_file_infor7rBFileTypeFileNotADirectoryErrorFileNotFoundErrorrsortedrrrr.raparquet_datasetr9 get_fragmentsrKrr:pyarrow.parquetparquetParquetDatasetrWrr empty_table to_pandasrp)rSrv dataset_infopath_normalizedrr9dataset_selectorr metadata_filerhdatasetfragfipqs r=rnz"ReadParquetPyarrowFS._dataset_infos34 42 4I ..  R I' #(#5#5dd#K   &*WW%:%:;K%L! ::)<)<<  ((JI}&&// < )  #M2//!&&#ww7; 23?F?T?T?V-7;OD$''2- [)/3Vc,{:S6T-T \*   *H$>$@$J$J$L [!#/ o #$56 RAPQ..t4QQIQ R(-*G. )sHAI=+:I8 % I= KK K  K8I==J>"J32J>=J>c|jr8|j,|jj}t|j|St dgt |jdzzdfS)aIf enabled, compute the divisions from the collected statistics. If divisions are possible to set, the second argument will be the argsort of the fragments such that the divisions are correct. Returns ------- divisions argsort NrZ)rrr_divisions_from_statisticsrrrr)rS index_names r=_division_from_statsz)ReadParquetPyarrowFS._division_from_stats6s]  # # (>J-d.H.H*U UdVs4#:#:;a?@A4GGr?c@td|jdDS)z:Whether all statistics have been fetched from remote storec3>K|]}t|tvywr6)r.r)rrs r=rz.Hs 05HUO| + sr)allrnrvs r=all_statistics_knownz)ReadParquetPyarrowFS.all_statistics_knownFs' 9=9K9KK9X   r?c |jdSrrrvs r=_fragment_sort_indexz)ReadParquetPyarrowFS._fragment_sort_indexL((++r?c |jdSrrrvs r=rzReadParquetPyarrowFS._divisionsOrr?cz|jdk\ryt|try|j|t|Sr)r|rr! substitute)rSrRs r=_tune_upzReadParquetPyarrowFS._tune_upRs8  * *a /  fn -   ~d';<#&88%@@r?c b|jj}|jj}|j|jj}|jdj }|r(|t |j}|j|t|tjtdtjt|j||j|j ||||j"|j$j'd|j(dS)Nrr)fragment_wrapperrWrrrGT)rrrGrH_data_producer)rcopyrrrnremove_metadatarrrr r_table_to_pandas_fragment_to_tablerKrrjrWrrrorH)rSrrrrrs r=_filtered_taskz#ReadParquetPyarrowFS._filtered_tasks,,##%ZZ__ :: !J##H-==? v||, NN: &  1 1 $77!0NN5)dgg"  " 00++///:$($@$@#  r?cXtt|tr |j}n|}t|trt j |}|j|||dtjjdtjtdtddS)NiTz4 MiBz 32.00 MiB)hole_size_limitrange_size_limit) pre_buffer cache_options)rrr batch_sizefragment_scan_options use_threads) rGrrKrTrrrto_tablerBrParquetFragmentScanOptions CacheOptionsr3)rrWrrrTs r=rz'ReadParquetPyarrowFS._fragment_to_tables! & 8'00H'H gt $..w7G  ""$**"G"G oo$/$8%0%=#H#)!  r?c &|i}n|j}|jdd|jd t|jdd|||j dd|j ddd|ddi}||j |}|S) Nignore_metadata types_mapper)rFrGrHrF self_destructT)rrrrF)rrrrIro set_index)tablerrrGrHrs r=rz%ReadParquetPyarrowFS._table_to_pandass  " O-224O-t4 U__  /"1"5"5nd"K+(? (++M5A)--otD    !    !j)B r?)r|dictr{)r|boolr)r8r~rr _defaultsr~rrrrjrrrrrr]rnrrrrrrrrr|r staticmethodrrrFr?r=rr;s`K$ $$#'#I 88  +%Z ( D  BBT KKZ H H ,,= ' 'UU*AA  @  >r?rceZdZdZgdZiddddddddddd d d d d dd ddddddddddddddidddd dddZedZdZedZ d%d Z ed!Z ed"Z d&d#Z ed$Zy)'ReadParquetFSSpeczRead a parquet dataset)r9rrWrrrrrmetadata_task_sizesplit_row_groups blocksizeaggregate_filesparquet_file_extensionrUrrrrVrp_pq_length_statsrNrWrrrrFrr!r"inferr#defaultr$r%)z.parqz.parquetz.pqrUfsspecrr3rrGrrV)rpr&c^|jd}t|tr t|S|S)Nr)r[rrr&)rS_engines r=rzReadParquetFSSpec.engines*,,x( gs #g& &r?c |jdS)Nr)_planrvs r=rzReadParquetFSSpec._divisionsszz+&&r?cD|jdx}r|Stdi|jxsi\}}}}|jj |j |j |||j\}}}}||d<t|t}d}|jd} | d}| rt| tr| g} n| } |j} |jdvrC|jdk(r6t|jd r|jj} nd } nd} |||j | |j"|j$|j| |j&|j(|j*|j,||d |f } |jj.| } g}g}| d rVt|j t0rt3d |j Dg}n-|j |j4zdzg}n| dj6}|D]"}|j9|j;|$t=|| d<|jj?| }| d} t| tr| gn| } tA|| d|\}} }|jBjDtFk(rd|jB_"|| d<| | d<|| d<|j"| d<t=| }|tHvr|jjK| \}}}tM||\}}tO||| \}}|j$r|rtQ|||j$\}}tS|| tU|}d}tU|dkrd}|g}d}tW||||||dtH|<tH|| d<| |_,| S)NrprrFrT)r'adaptiver(default_blocksize128MiB)readrhas_metadata_filec3DK|]}|jds|yw)rN)r)rr9s r=rz2ReadParquetFSSpec._dataset_info..gsR$t}}[7QRs  rrrhru all_columnsrr{)rparts statisticsr common_kwargsplanrF)-r[r)rrrr9rUrrr2rrr#r"rr0rrrWr$rr!r%_collect_dataset_infornextseprrrhr._create_dd_metar'rrrr_construct_collection_plan_align_statistics_aggregate_row_groupsr%_calculate_divisionsrrrp)rSrrr`r other_optionsrjpathsauto_index_allowed index_operandrr#argsrrhfiles_for_checksumfilerr5 dataset_tokenr7rr9rrs r=rnzReadParquetFSSpec._dataset_info s34 42 4I 64;;#4" 6     9= 8V8V II OO    9 5E?$5-> ()u"23" W-  !%  Z s;"OE!ENN  $9 9~~*4;;(;< $ = = ?I (II   OO   $ $ LL  ! !    % %  # #  ' '$*    &9t{{88$?  + ,$))T*R$))RR&"'+ii"&&&8;&F%G"!-d!3!9!9 & /D OOBKK- .  / $,H#5 Z {{**<8W%%eS1u#4 %1$  e[ ::??j ("DJJO$( [! % W&1 ]#.2.F.F *+ .  ,*.++*P*P+ 'E5- -UE:LE51|LLE5||,UE4<<H u-UL#e*MIE9~!(   /#&!. +L ' ,M: V#/ r?ct||j|jd|d}|jr)t|tj ||j dS|S)Nr7T)rr)r _io_funcr-rVoperatorgetitemr)rSrrtsks r=rz ReadParquetFSSpec._filtered_tasksP4 7(;E(BSWX <<h..T\\!_E E r?c |jdrtS|j}t|j|d|d|j |d|ddi|jdS)NrrjrurrrGr9)r-r rnr"rr)rSrs r=rLzReadParquetFSSpec._io_funcsm ::g O)) % KK    % LL  !  "? 3  JJ '  r?c |jdS)Nr:rmrvs r=r-zReadParquetFSSpec._plans!!&))r?cljs'tfdtjDSy)z7Return known partition lengths using parquet statisticsc3\K|]#\}}jr|jvr|%ywr6 _filteredr)rr,lengthrSs r=rz1ReadParquetFSSpec._get_lengths..s1Av~~d.>.>)>s),N)rWr enumerater&rvs`r=r]zReadParquetFSSpec._get_lengthss4||!*4+@+@!A  r?cjdr*tfdtjdDStdtDS)z6Ensure that partition-length statistics are up to dater8c3bK|]&\}}jr|jvr|d(ywnum-rowsNrT)rr,statrSs r=rz5ReadParquetFSSpec._pq_length_stats..s6At~~d.>.>)>Z s,/c3&K|] }|d ywrZrF)rr\s r=rz5ReadParquetFSSpec._pq_length_stats..sSdj)Sr)r-rrW_collect_pq_statisticsrvs`r=r&z"ReadParquetFSSpec._pq_length_statssR ::l #(L)AB S6LT6RSS Sr?r)r|z tuple | None)r8r~r__doc__rrrrrrnrrLr-r]rr&rFr?r=r r sf K,44 d   4  u   d G Y 4 !"> h ) ?D) t!" 5#$ $ 'I, 'FFP     ** T Tr?r c|dk(r td|Q| t|jdddk(s#tjj dddk(r ddlm}|}|Sd }|S) N fastparquetz#Fastparquet engine is not supportedrrcudfzdataframe.backendpandas) CudfEnginer3)rwr4splitr rrodask_cudf.io.parquetrd)rrrds r=rrso !"GHH ~  $!5!5c!:1!=!G [[__0( ;v E 7F MF Mr?c |rt|t|k7rg}|rCttt||Dcgc]\}}|ddkDr||fc}}}|xsggg\}}||fScc}}w)Nr[r)rrzip)r7r8partrresults r=r@r@sc%jC O3  (+5*'=#eZ(1,5M  #.r2hz * sA%c|ddur|dnd}|d}|d}|d}|r%|s|r!t|dkDrt||||||\}}||fS)Nr"Tr#rjaggregation_depthrZ)rr$)r7r8rr#r"rjrls r=rArAs&22D%E%M [!SW$$67 d B$%89 )c2B.Ca.G 4z9.>DU! E: * r?cd}|rc|jddrQ|jdd}|d}|rt|dk(r|nd}|dur"|r t||D]}|d|vs |d}n|xsd |dzzS) Ngather_statisticsFrrrZrrrr6)rorr()r8rrrrrprocess_columnssorted_column_infos r=rBrBsIl&&':EB*../DdKW%#(SZ1_%$ u ,/&4O' "&f-6 2; ?I    3;?33r?ceZdZUdZGddeZGddeZded<ddZdd Z dd Z e dd Z dd Z e dd Zy)rOz/Manage filters in Disjunctive Normal Form (DNF)ceZdZdZddZy)_DNF._OrzFrozen set of disjunctionscd}|Dcgc],}t|dr||jn||.c}Scc}w)Nctt|tr&|r$t|dttfr t|S|gSr)rrr)vals r= _maybe_listz+_DNF._Or.to_list_tuple.._maybe_list6s1c5)cjQ%QU6W9$u r?r\rr\)rSrwrvs r=r\z_DNF._Or.to_list_tuple4sR   sO4  1 1 34$S)* s1<Nr|rr8r~rr_r\rFr?r=_Orrs1s ( r?r{ceZdZdZddZy) _DNF._AndzFrozen set of conjunctionsc&td|DS)Nc3XK|]"}t|dr|jn|$yw)r\Nrx)rrvs r=rz*_DNF._And.to_list_tuple..Is/(/sO'D!!##Ms(*)rrvs r=r\z_DNF._And.to_list_tupleGs r?NryrzrFr?r=_Andr}Ds ( r?rz_And | _Or | NonerQc0|j||_yr6) normalizerQ)rSrWs r=rXz _DNF.__init__Psw/ r?c6|jjSr6)rQr\rvs r=r\z_DNF.to_list_tupleSs}}**,,r?c,t|jSr6)rrQrvs r=__bool__z _DNF.__bool__VsDMM""r?c|sd}|St|trHt|dtr|n|g}j|Dcgc]}j|c}}|St|trBt|dtr t djj|ff}|St|jrjfd|D}|St|jrpg}t j|Dcgc]}j|c}D])}|jjd|D+j|}|St t|dcc}wcc}w)z9Convert raw filters to the `_Or(_And)` DNF representationNrz0filters must be List[Tuple] or List[List[Tuple]]c3NK|]}j|D]}|ywr6)r)reseclss r=rz!_DNF.normalize..fs&LA3==;KLRRLRLs"%c3.K|] }|D]}|ywr6rF)rrrs r=rz!_DNF.normalize..js%@Qa%@b%@b%@sz not a supported type for _DNF) rrr{rrr itertoolsproductrrr7)rrWrj conjunctions conjunctiontotalrr's` r=rz_DNF.normalizeYslF" ! &&0T&B7 LWW|T chh{3TUF  ''!*e, RSSWWchhz245F  )WWLLLF  *E&&7(Kaq)9(KL B SXX%@!%@@A BWWU^F tG}o-KLM MU)Ls FFcLt|ts t|}t|tsJ|j|j}t|S|j|j}t|S|j|j|jg}t|S)z Combine with another _DNF object)rrOrQr)rSotherrjs r=rZz _DNF.combineps%&KE%&&& == ^^F F| ^^ #]]FF|YY u~~>?FF|r?c d}t|ttttt t frt|jtst|jtrn|jjj|jk(rA|j}|jjd}|j}|||f}t-|St|jtsmt|jtrR|jjj|jk(r$tttttttti}|}|j||j}|jjd}|j}|||f}t-|St|t t"fr|j%||jj&}|j%||jj&} |rC| rAt|t r|j)|| g}t-|S|j+|| g}t-|Sr)rrrrrr rrightrleftrrr_operator_reprrrorrrPrQrr{rO) rpq_exprpredicate_exprrQrrcolumnvaluefliprrs r=rPz_DNF.extract_pq_filters}s nr2r2r2&> ?~33T:~22J?"''--33w}}D#22',,44Q7&,,"B..H~+~22D9~22J?"''--33w}}DBBBB7#XXb"%44'--55a8&++"B.H~b 2))'>3F3FGPPD**7N4H4HIRREnc2"xxu 6HH~ #wwe}5HH~r?N)rW _And | _Or | list | tuple | Noner|r}ryr)rWr)rz'_DNF | _And | _Or | list | tuple | Noner|rO)rrrrr|rO)r8r~rr_ frozensetr{r__annotations__rXr\r classmethodrrZrPrFr?r=rOrO.sa9i&y 0-#, ""r?rOc|rt|tstdt|d|jj j ht|jz}t|j|std|d||jdrgS|jj}t|jdDcgc]!\}}|jr||jvr|#}}}t!t#|rdnd}|rt%t}|D]D}t|t&r|gn|D]*} | j)d d } || j+| ,Ft|j-} t/t0} t3j4t7d t9| |D cgc]=}| tt;j<| |||zD cgc]} ||  c} || ?c} }d }tt;j<|St1||| Scc}}wcc} wcc} }w) z+Collect Parquet statistic for dataset pathsz#Expected columns to be a list, got rzcolumns=z must be a subset of rr7Fpiecerr)rrrr7rrrrrissubsetr-rLrjrWrUrrr*rrrorrr+_read_partition_stats_groupr rr rrchain)rrallowedrjr,rir7parallelgroupspr9 group_keysfunckrjs r=r^r^s* '4(B4=/QRST T::##(()C ,==7|$$W-xy0EgYOP P zz'    B!G!45  At~~d&6&6!6  E L,5"5H T" 'D)$5dV4 'uuW~a(t ##A& ' '&++-( 23q#j/8< !1;AH 1MNAfQiN #      IOOV,--+5"gFFO 6O s&H1;%H< H7,H< 7H< cFdd}|Dcgc] }||||c}Scc}w)z)Parse the statistics for a group of filesct|ts|g}i}d}|xsg}|D]}|d}|d}|ddgk(rdn|d}|j|d5} tj| j } ddd|tt  j}|D]}  j| } || jz }t | jD]} | j| }|j}||vs%|js2|jjsI||vrYt||d|jj||d<t!||d|jj ||d<|jj|jj d||<|j#Dcgc]}|||d||dd }}||d S#1swYuxYwcc}w) NrrrZnone) default_cacheminrzrrz)rrrz)r[r)rropenr ParquetFilemetadatar r row_groupr num_columnsrrr8 has_min_maxrrzr)rirjr column_statsrrrr9 row_groupsfmdrgrr,r_rcolumn_stats_lists r=_read_partition_statsz:_read_partition_stats_group.._read_partition_statss$%6D -R "AgJE8D!&qdV!3qJV4 0^^A&// 0!!%(9(9":;   "LL, I...y445"A#**1-C--Dw>>cnn.H.H#|3>+=+=+.>>+=+=6" T 2" " "J%))+   #D)%0#D)%0   %1BCCG 0 06 s G$G1$G. rr6rF)r7rjrrris r=rrs+1DhJO O !$G < OO Osct|ttfs|g}g}|D]T}ddg}|D])}|j|d}t |dkDs$|d}n|j |j dV|S)Nz://z::rZr)rrrrerrr)r9rjrprotocol_separatorsr=res r=rrs dT5M *v F %$dm& CGGCOE5zA~!H    ahhsm$% Mr?cd}|d}t|dD]\}}|d|k(s|}ntd|dd}g}|D]1}|d|dd} |d|dd } |j| | f3g} tj|}|j } || } | j stdgt|d zzdfS| D]\} } | j| | }| j|t| | fS) NrLrrrz Index column z not found in statisticsr8rrzrZ) rWrrr4Seriesris_monotonic_increasingrr)aggregated_statsrcol_ixpeak_rgixr_last_maxminmax file_statsfile_minfile_maxrr sorted_minmaxs r=rr.sZ Fq!GWY/0 C  J .F  J<'? @  H F&, i(0>uEi(0>uE x*+ , I YYv FnnG7OM  0 0dVs#34q89:D@@+("X  W $$r?cVgd}gd}gd}gd}i}|D] }||||< gx|d<}|dD]x}i} |j| |D] }||| |< g| d<|dD]G} i} | dj| |D] }| || |< i| d<| d3|D]}| d|| d|<Iz|S)zTake the raw file statistics as returned by pyarrow (as a dict) and filter it to what we care about. The full stats are a bit too verbose and we don't need all of it.)rrr)rtotal_byte_sizesorting_columns) num_valuestotal_compressed_sizerr)rrz null_countrdistinct_countrrr8)r) originalfile_level_statsrg_statscol_meta col_statsr/rrgsrrg_outr_col_outs r=_extract_statsrNs1 IH H I C #TND #  C |$F 6 $Dd8F4L $yi= FCG 9  $ $W -  * #D   *$&GL !< (! F.1,.?.E %d+ F F F" Jr?ci}|D]9}|jD]$\}}||vr|g||<||j|&;i}|jD]$\}}|j|}|s||||<&|Sr6)itemsrro)dicts agg_funcsrjdrvresult2aggs r= _agg_dictsrs F $GGI $DAqCq q   #  $$ G  1mmA QGAJ  Nr?cg}d} g}|j| |D]}|j|| |dz }6#t$r|jYnwxYw|Dcgc]}t||c}Scc}w)NrrZ)r IndexErrorrr)colsagg_colsrZr,innerr_r's r=_aggregate_columnsrsG A u  % SV$ % Q    KKM  .5 5Jq( # 55 5s<AAA5c`|Dcgc]}|| }}t|dkDr||SdScc}wr)r)rrys r=_get_min_max_valuers7'q'A'!fqj47*d* (s++c4ddd}tttt|dd}tttt|d}g}|D]M}|j }|j ||j t|jd |O|S) z'Aggregate RG information to file level.c"t|tSr6)rrrs r=rz/_aggregate_statistics_to_file..+As3r?c"t|tSr6)rrzrs r=rz/_aggregate_statistics_to_file..rr?r)rc4t|jSr6rrrs r=rz/_aggregate_statistics_to_file..CFJJLr?)rrr8rr)rrrr)r^rrrrrupdater)r agg_statsragg_funcr file_stats r=rrs43I "%#&jI>0 H-AH L NN$  *IMM,$?JK L r?ctjdtjtfd|DddS)Nct|dtr|dj|d<|dt|djj fS)NrZr)rrKrTrrto_dict)token_fragments r=_collect_statisticsz/_gather_statistics.._collect_statisticssO nQ' 9 .q 1 : :N1 a .1B1K1K1S1S1U"VVVr?c3.K|] }|ywr6rF)rrrs r=rz%_gather_statistics..s 94  & 9s threading) schedulerr)r r+rr)rrs @r=_gather_statisticsrsF \\WW << 95 99[   r?cg}t||D],\}}t|x}tvs|j||f.tj j d|Dcgc] }t|c}Scc}w)zHCollect statistics for a list of files and their corresponding fragments)rhr.rrr itertoolz partition_allr) file_infosr to_collectrrrbatchs r=rrs~J:y1- te_ $E\ 9   udm ,- __222zB   5!  s A5c  tjtjdd}tt|tjtjtjtjt t |dS)zCombine multiple file-level statistics into a single dict of metrics that represent the average values of the parquet statisticsc4t|jSr6rrs r=rz _combine_stats..rr?)rrrr)rrrrr)r8meanrrrr)rrs r=rrs_",#-??0H %e,"(oo))1HE    r?)snappyTFFFNNNNTNr'NNNr{r6)rrrz list | Noner|zlist[dict] | None) __future__rrrrMrDrer8rrabcr collectionsr functoolsrrnumpyrrcr4r3rBpyarrow.computerr]pyarrow.datasetrra pyarrow.fsrjrrrrtlzr fsspec.utilsrr r dask._task_specr r dask.corer dask.dataframe.dask_expr._exprr rrrrrrrrrrrrrrr$dask.dataframe.dask_expr._reductionsrdask.dataframe.dask_expr._utilrdask.dataframe.dask_expr.iorr dask.dataframe.dask_expr.io.ior!dask.dataframe.io.parquet.corer"r#r$r%r&r'r(dask.dataframe.io.parquet.utilsr)dask.dataframe.io.utilsr* dask.delayedr+ dask.tokenizer,r-r. dask.typingr/ dask.utilsr0r1r2r3r4registerFileInfor>rArGrint8 Int8Dtypeint16 Int16Dtypeint32 Int32Dtypeint64 Int64Dtypeuint8 UInt8Dtypeuint16 UInt16Dtypeuint32 UInt32Dtypeuint64 UInt64Dtypebool_ BooleanDtyperr5float32 Float32Dtypefloat64 Float64DtyperErrrrKrDatasetr FileFormatrSchemar ParquetSchemar FileMetaDatarrrrr0rIrrr rr@rArBrOr^rrrrrrrrrrrrFr?r=rFs" #.' )$5;G9@0 LLOO"%%..)*  BGGI|r||~ BHHJ   BHHJ   BHHJ   BHHJ   BIIK! BIIK! BIIK! BHHJ! BIIK!BJJL/"//#BJJL/"//# "#  VJVJr% %--(!)!%**+,"))$%"**+",""//* + $  >VIV.StSD    #K\ ,$^j $kj Zi;iX sT sTv"&"4.rrv/39G 9G +9G9Gx7Pt  %@.b  6 + 6     r?