&`ivddlZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z m Z mZmZmZddlZddlmZddlZddlmZddlmZmZddlmZddlmZdd l m!Z!dd l"m#Z#m$Z$m%Z%m&Z&dd l'm(Z(m)Z)m*Z*dd l+m,Z,dd l-m.Z.ddl/m0Z0ddl1m2Z2ddl3m4Z4m5Z5m6Z6ddl7m8Z8m9Z9m:Z:m;Z;ddlm?Z?m@Z@mAZAddlBmCZCer ddlDZDddlEmFZFejGeHZIedZJdZKdZLdZMdZNdZOdZPdZQdZRdZSGd d!ZTd"ZUeGd#d$ZVd%e@d&eWd'eVfd(ZXGd)d*e.ZY dWd+ee(gee(fd,e eZefd-ee[d.ee eZd/ee eZeZfd&ee eZd0eee\d1fd2e eTd3e]d4e9d5ed6d'e d7fd8Z^ddd9d9dd:d;dee[d?e]d@e]d,ee eZefd'e d7fdAZ_d;dc|jSr8)r9rAs r<originalz_ParquetFragment.original|s ~r>ctj|jj|jj|jj|jj|jffSr8)r4 make_fragmentr9formatpath filesystempartition_expressionr:rAs r< __reduce__z_ParquetFragment.__reduce__s<- N ! N  N % N / O 0   r>cP||||}t||Sr8)rEr4)rFrGrHrIr6fragments r<rEz_ParquetFragment.make_fragments*''j:NOO)444r>N)r?r-) __name__ __module__ __qualname____doc__intr=propertyr6rCrJ staticmethodrEr>r<r4r4os**$/$C$$$$3XX   55\555r>r4cddl}t|j|jD]B\}}t ||jr(t ||jrtd|dCdS)aXCheck for the legacy tensor extension type and raise an error if found. Ray Data uses an extension type to represent tensors in Arrow tables. Previously, the extension type extended `PyExtensionType`. However, this base type can expose users to arbitrary code execution. To prevent this, we don't load the type by default. rNz,Ray Data couldn't infer the type of column 'aN'. This might mean you're trying to read data written with an older version of Ray. Reading data written with older versions of Ray might expose you to arbitrary code execution. To try reading the data anyway, set `RAY_DATA_AUTOLOAD_PYEXTENSIONTYPE=1` on *all* nodes.To learn more, see https://github.com/ray-project/ray/issues/41314.)pyarrowzipnamestypes isinstanceUnknownExtensionTypePyExtensionType RuntimeError)schemapanametypes r<check_for_legacy_tensor_typerbs&, 55   d dB3 4 4  "$: : VtVVV   r>c@eZdZUdZeeed<eeed<dS)_SplitPredicateResulta{Result of splitting a predicate by column type. Attributes: data_predicate: Expression containing only data column predicates (for PyArrow pushdown), or None if no data predicates exist. partition_predicate: Expression containing only partition column predicates (for partition pruning), or None if no partition predicates exist. data_predicatepartition_predicateN)rMrNrOrPr r*__annotations__rTr>r<rdrds>TN"""!$'''''r>rd predicatepartition_columnsr?cRtt|}||z }||z}|st|dS|std|St|tr|jt jkrt|j |}t|j |}dttdttdttfd}||j |j }||j|j} t|| StddS)aSplit a predicate into data-only and partition-only parts. This function extracts both data column predicates and partition column predicates from AND chains, enabling both PyArrow pushdown (data part) and partition pruning (partition part). Args: predicate: The predicate expression to analyze. partition_columns: Set of partition column names. Returns: _SplitPredicateResult containing: - data_predicate: Expression with only data columns (for PyArrow pushdown), or None if no data predicates can be extracted. - partition_predicate: Expression with only partition columns (for pruning), or None if no partition predicates can be extracted. Examples: >>> from ray.data.expressions import col >>> # Pure data predicate: >>> result = _split_predicate_by_columns(col("data1") > 5, {"partition_col"}) >>> result.data_predicate is not None # Should have data predicate True >>> result.partition_predicate is None # Should not have partition predicate True >>> # Pure partition predicate: >>> result = _split_predicate_by_columns(col("partition_col") == "US", {"partition_col"}) >>> result.data_predicate is None # Should not have data predicate True >>> result.partition_predicate is not None # Should have partition predicate True >>> # Mixed AND - can split both parts: >>> result = _split_predicate_by_columns( ... (col("data1") > 5) & (col("partition_col") == "US"), ... {"partition_col"} ... ) >>> result.data_predicate is not None # Should have data predicate True >>> result.partition_predicate is not None # Should have partition predicate True >>> # Mixed OR - can't split safely: >>> result = _split_predicate_by_columns( ... (col("data1") > 5) | (col("partition_col") == "US"), ... {"partition_col"} ... ) >>> result.data_predicate is None # Should not have data predicate True >>> result.partition_predicate is None # Should not have partition predicate True N)rerfleftrightr?c|r|r||zS|p|Sr8rT)rkrls r<combine_predicatesz7_split_predicate_by_columns..combine_predicatess' $ $e|#=5 r>)setrrdrZr)opr+AND_split_predicate_by_columnsrkrlr r*rerf) rhrireferenced_cols data_colspartition_cols_in_predicate left_result right_resultrnrerfs r<rrrrs\r/ ::;;O"33I"14E"E &Y$ISWXXXX Y$DiXXXX)Z(( Y\Y]-J-J1).BSTT 29?DUVV  !4. !)1$ ! d^ ! ! ! !,+  & (C  10  +\-M  %)?R    !$ O O OOr>ceZdZdZdgZddddddddeddddd deeeefde eed e e ee fd e e ee fd e e e ge fd e d de eedfde edede edeeddfdede eeffdZdefdZ d.dede edeefdZdZedefdZdefdZdefd Zde eefd!Zde eefd"Zde eefd#Zd$e ddffd% Z!d&ee"defd'Z#e$d(e d)d*d)d+e d)d,e eedd)f d-Z%xZ&S)/ParquetDatasourcea;Parquet datasource, for reading and writing Parquet files. This implementation uses PyArrow's `ParquetDataset` abstraction for dataset reads, and thus offers automatic Arrow dataset schema inference and row count collection at the cost of some potential performance and/or compatibility penalties. parquetNhiveF) columnsdataset_kwargsto_batch_kwargs _block_udfrHr^ meta_providerpartition_filter partitioningshuffle include_pathsfile_extensionspathsr|r}r~rrHzpyarrow.fs.FileSystemr^pyarrow.lib.Schemarrrrfilesrrc Httt| |_|js7t jjjrtdd|_ |js;ddl m }|t j d|_ ||_t!||\}|_t%j|jt)jj}t/||| | }|r t1|\}}ngg}}|t2dni}d |vrtd d|d <t7t9|||}d \}}|+|jr t=||jd| \}}ngg}}|i}d t1|j|D|_d |jD|_ ||_!||_"| |d|_#nSi|_#|$|j#$d|D|$|j#$d|D|}|V| T|jrMtK| }||jdj&}|r!t9|'}||ng|_(|duotS|j(dk|_*||_+|j,|_-t]| |j |_/d|_0| |_1| |_2| dkr$tfj45|_0n>tm| tnr)tfj45| j8|_0ts|j}tu||;||j }ty|||_=t}|t)jj?|_@dS)NzBecause you're using Ray Client, read tasks scheduled on the Ray cluster can't access your local files. To fix this issue, store files in cloud storage or a distributed filesystem like NFS.r)NodeAffinitySchedulingStrategyF)softretryable_errors)rrz|Please note that `ParquetDatasource.__init__`s `dataset_kwargs` is a deprecated parameter and will be removed in the future.rzuThe 'partitioning' parameter isn't supported in 'dataset_kwargs'. Use the top-level 'partitioning' parameter instead.)NNc4g|]\}}t||SrT)r4).0rLr6s r< z.ParquetDatasource.__init__..s6   #) Xy 1 1   r>cg|] }|j SrT)rG)rps r<rz.ParquetDatasource.__init__..s:::Q!&:::r>ci|]}||SrTrTrcols r< z.ParquetDatasource.__init__..s,N,N,N#S#,N,N,Nr>ci|]}||SrTrTrs r<rz.ParquetDatasource.__init__..s,S,S,S#S#,S,S,Sr>r)r|r^local_scheduling)Asuperr=rr_supports_distributed_readsrayutilclient is_connected ValueError_local_schedulingray.util.scheduling_strategiesrget_runtime_context get_node_id _source_pathsr( _filesystemrwrapr get_currentretried_io_errorsr#rWloggerwarningget_parquet_datasetlist fragments!_infer_data_and_partition_columns _pq_fragments _pq_pathsr_to_batches_kwargs_projection_mapupdater'rGkeys_partition_columnslen_partition_columns_selected _read_schemar^ _file_schema_get_partition_columns_schema_partition_schema_file_metadata_shuffler_include_paths _partitioningnprandom default_rngrZr seed_sample_fragments_fetch_file_infos_get_data_columns_estimate_files_encoding_ratio_encoding_ratio_estimate_reader_batch_sizetarget_max_block_size_default_batch_size)r;rr|r}r~rrHr^rrrrrrr listed_files file_sizespq_ds data_columnsriactual_partition_columnsrparsed_partitionssampled_fragmentssampled_file_infos __class__s r<r=zParquetDatasource.__init__#s"    /?/F/F+F(/ CHO4G4T4T4V4V O  "&/  U U U U U U%C%C'))5577e&&&D "#"?z"R"Rt).  (466H   #  -+      ' #\ 2 E:: "B:E  % NNO      N ^ + +F *.~&$DKK^LL+5' '   92SU_Q/33/ // 35b/  " O   '*5?J'G'G   ;:%/:::$"1  $5$=#'D #%D '$++,N,N,N,N,NOOO ,$++,S,SAR,S,S,STTT$5  $)Aeo)A' 55E %eoa&8&= > >   J+/0A0F0F0H0H+I+I( )A(L $ $RT  T ) Nc$2I.J.JQ.N (#!L!> $." " (,$+) g  +-9+@+@+B+BD ( ( !2 3 3 O+-9+@+@+N+ND (.    / **,,!3     >     $?  7 9 9 O$ $    r>r?cP|jikrdS||jSNr)r_estimate_in_mem_sizerrAs r<estimate_inmemory_data_sizez-ParquetDatasource.estimate_inmemory_data_sizes,  2 % %1))$*<===r> parallelismper_task_row_limitcB |jtt|j|jfd|jt D}tttt|\}}n|j|j}}||j |j |j | |j }g}|j|jndtt!j||t!j||D]\}} t |dkrt%d||| d} |j |j|j||||j |j|jf \  |t9|f f d | |||S)Nc g|] }| SrTrT)rifiles_metadatas r<rz4ParquetDatasource.get_read_tasks..s.'''q!'''r>) file_schemapartition_schemaprojected_columnsrr)num_rows size_bytes input_files exec_statsc 6 t  | Sr8)read_fragments) r5 block_udfrdata_columns_rename_mapdefault_read_batch_size_rows filter_exprrrir read_schemato_batches_kwargss r<z2ParquetDatasource.get_read_tasks.."s2!)4$/)#%$# ) )r>)r^r)rrrWrr permutationrmap_derive_schemarrrget_current_projectionr_predicate_expr to_pyarrowr array_splitrrrrrget_column_renames_get_partition_columnsrrappendr)r;rrshuffled_files_metadata pq_fragmentspq_paths target_schema read_tasksrrmetarrrrrrrrirrrs @@@@@@@@@@@r<get_read_tasksz ParquetDatasource.get_read_taskss  ' 3!#d&8$."I"IJJN''''5AA#nBUBUVV''' #&*#dC9P4Q*R*R%S%S "L(("#L ++  )!3"99;; ,   #/  + + - - -  !$ N< 5 5 N8[ 1 1! ! 7 7  Iu9~~"" 55i@@! D$'(&&((''))++--!#"  !,'!   &              ('9!    *r>cdS)zuReturn a human-readable name for this datasource. This will be used as the names of the read tasks. ParquetrTrAs r<get_namezParquetDatasource.get_name7s yr>c|jSr8)rrAs r<supports_distributed_readsz,ParquetDatasource.supports_distributed_reads>s //r>cdSNTrTrAs r<supports_projection_pushdownz.ParquetDatasource.supports_projection_pushdownBtr>cdSrrTrAs r<supports_predicate_pushdownz-ParquetDatasource.supports_predicate_pushdownErr>cp|}|}||dS|pg|pgzS)zBOverride to include partition columns in addition to data columns.N)rr)r;rris r<rz(ParquetDatasource.get_current_projectionHsN--//  7799  $5$=4"'8'>B??r>c~jdSjsdSfdjD}|r|SgS)auExtract partition columns from projection map. This method extracts partition columns from _projection_map, which is the source of truth after projection pushdown. Since partition columns are now included in _projection_map during initialization when requested, we can reliably extract them from the map. Returns: List of partition column names in the projection, None if there's no projection (meaning include all partition columns), or [] if partition columns aren't in the projection map (meaning include no partition columns). Nc&g|] }|jv |SrT)r)rrr;s r<rz.hs-   #AX:X:XC:X:X:Xr>rrr)r;partition_colss` r<rz(ParquetDatasource._get_partition_columnsSsq   '4& 4    /4466     "! !  r>cr|jdS|jfd|jD}|S)aExtract data columns from projection map, excluding partition columns. Partition columns aren't in the physical file schema, so they must be filtered out before passing to PyArrow's to_batches(). Returns: List of data column names to read from files, or None if no projection. Can return empty list if only partition columns are projected. Ncg|]}|v| SrTrT)rrrs r<rz7ParquetDatasource._get_data_columns..s*   #^:S:SC:S:S:Sr>r )r;rtrs @r<rz#ParquetDatasource._get_data_columnsus\   '40    /4466   r>predicate_exprcbt|j}|s!t|St ||}|j|jt|j}g}g}t|j |j D]J\}}| ||jr*| || |K||_ ||_ ddl } | |} |j-tt| |jS| S)u^Apply a predicate with data pushdown and partition pruning. This method optimizes predicates in three ways: 1. Data predicates → pushed to PyArrow (row-level filtering) 2. Partition predicates → used for partition pruning (file-level filtering) 3. Mixed predicates → both optimizations applied together Nr)rorrapply_predicaterrrfrr'rWrrevaluate_predicate_on_partitionrcopyrery) r;rr split_resultparserpruned_fragments pruned_pathsrLrGr datasourcers r<rz!ParquetDatasource.apply_predicatesIT455 ;77**>:: :3>>RR   , 8".();<rcbtd|D|jz}t|S)Ncg|] }|j SrTr6rr5s r<rz;ParquetDatasource._estimate_in_mem_size..s:::11;:::r>)sumrround)r;r in_mem_sizes r<rz'ParquetDatasource._estimate_in_mem_sizes4:: :::;;d>RR [!!!r>rpyarrow.Schemarrrc ddl}|| nMt}|t|ng}||fd|Dzj |'| fd|D j |j } ||j j n,#t $rtddYnwxYwt  S) z(Derives target schema for read operationrNcPg|]"}|jdk |#S))get_field_indexr`)rr5rs r<rz4ParquetDatasource._derive_schema..sC'66qv>>"DD EDDr>)fieldsmetadatac:g|]}|SrT)field)rcolumnrs r<rz4ParquetDatasource._derive_schema..s'MMM$$V,,MMMr>zdFailed to infer schema of dataset by passing dummy table through UDF due to the following exception:T)exc_info) rVrr^r( empty_table with_metadata Exceptionrdebugrb) rrrrrr_file_schema_fieldspartition_schema_fields dummy_tablers ` @r<rz ParquetDatasource._derive_schemas|   "'MM!%k!2!2 *:*F%&&&B $ II&!8%-&  M  (IIMMMM;LMMM&M  !'3355K  * ; 7 7 > L L!*!!     B!  %]333s(C&C+*C+r8)'rMrNrOrP_FILE_EXTENSIONSr%rstrr r rrrrrar!r&r boolr=rQrrrrrRrrr rrrr*rr4rrSr __classcell__)rs@r<ryrys"{ (,37489=8<>B8<04/;|F/C/C15#/3l l l S$s)^$l $s)$ l !c3h0 l "$sCx.1 l Xugun56l 45l t%99:;l  45l .l |,l ww'-.l l "$s),l l l l l l \>S>>>>EI]]]4ryrrrrrr^rrrrrzpyarrow.dataset.Expressionz pyarrow.Tablec # Kddlm} t|dksJtdt|d|D]jt jj} t  f dd| j D]!} | j dkr||| V| V"kdS)Nr)ArrowTensorTypezReading z parquet fragmentsc @ tj  S)N) r^rrrir include_pathr batch_sizer)_read_batches_fromrC) rrrrrLrrirr^rs r<rz read_fragments.. s8&!)(?"3)*'7"3   r>zreading batches)match) $ray.data.extensions.tensor_extensionr9rrr0rdatarrrrr)rrrrrrir^rrrrr9ctxtablerLs `````` ``` @r<rrs:EDDDDD y>>A     LL>C NN>>>???  h"..00'              '     E"~!!(#)E******KKK+   r>F)rr<r; use_threadsrrLr-r"r<r;rCc #v Kddlddlm} t pi d  dd} | | n| z| d|t ||d  f d } | | |Ed{VdS) z3Get an iterable of batches from a parquet fragment.rN)"_DatasourceProjectionPushdownMixinrCfilterr<r?pa.Tablec 3 K jd  d D]} j|g} r-tj|dj} rt |}|jdkr>|j dkr3| t |j }|VdS# j j$rp}t|}d|vrWUt jj}t%jj}t+dd|d|d}~wwxYw) z4Inner generator that yields tables without renaming.)r|rFr^rCrGrzNo match for FieldRef.NameNzFilter expression: 'z' failed on parquet file: 'z' with columns: rT) to_batchesTable from_batchesr for_block fill_columnrG_add_partitions_to_table num_columnsr append_columnrnullslib ArrowInvalidr5osbasenamerophysical_schemarXr])batchrBe error_messagefilename file_columnsrrrLr;r_partition_col_valuesr^rrCs r<_generate_tablesz,_read_batches_from.._generate_tablesas4 ,,$"'  $ % % --ug66.8??KK E(R45I5QQE"$))enq.@.@!//V>VE K% % Nv"   FFM, ==+7++HM::"8#;#ABB "G;GG&GG8DGG  sB>CEA+D??E)r?rG)rVray.data.datasource.datasourcerEdictpop setdefault_parse_partition_column_values_apply_rename_to_tables)rLr^rrrirrr<r;rCrrEfilter_from_kwargsr]r_r\s``` ` ``` @@r<r=r=7sd QQQQQQ.4"55$'' {CCK*..x>>%"  11  $$\:>>>9#\66666666666666r2II3r>ci}|t|}||j} fd|D}|S)Nc$i|] \}}|v || SrTrT)r field_namevalueris r<rz2_parse_partition_column_values..s5   ! E... ...r>)r'rGitems)rLrir partitionsrs ` r<rbrbsr J#L11U8=)) $    %/%5%5%7%7   r>r|_ParquetFileInfoc|jj}|jdkrdS|jdg}t t |jjtd}i}ttkrd|d<|j d|||d|}d}|D]0} | jdkr#tj | j | jz }n1t||S)Nr) row_group_idsr0batch_readahead)r|r^r<)avg_row_in_mem_bytesr(rT)rCr(num_row_groupssubsetmaxminr(PARQUET_ENCODING_RATIO_ESTIMATE_NUM_ROWSr MIN_PYARROW_TO_BATCHES_READAHEADrImathceilnbytesrk) rLr|r^r(row_group_fragmentr<r batches_iter avg_row_sizerWs r<_fetch_parquet_file_infor|s+ )H!##t"*111DD  ' 0 4   J @@@/0+,0%0  L#'L >A  9U\EN%BCCL E  )   r>cHeZdZUeeed<ded<deefdZdS)rkrozpyarrow._parquet.FileMetaDatar(r?c<|jdS|j|jjzSr8)ror(rrAs r<estimate_in_memory_bytesz)_ParquetFileInfo.estimate_in_memory_bytess#  $ ,4(4=+AAAr>N)rMrNrOr rQrgrrTr>r<rkrksZ#3-'''----B(3-BBBBBBr> file_infosctjjstSt |t |ksJd|D}d|D}dt ||D}|stSt j|}t d|ddt|tS)zReturn an estimate of the Parquet files encoding ratio. To avoid OOMs, it is safer to return an over-estimate than an underestimate. c>g|]}||ndSr8)r)rfis r<rz2_estimate_files_encoding_ratio..s:!!!FH##%%%T!!!r>cg|] }|j SrTrrs r<rz2_estimate_files_encoding_ratio.. s444QQ[444r>cHg|]\}}|dk | t||z  S)r)float)rr!r6s r<rz2_estimate_files_encoding_ratio.. sA!!! "K q==[4 kY&444r>z$Estimated parquet encoding ratio is z.3f.) rrdecoding_size_estimation'PARQUET_ENCODING_RATIO_ESTIMATE_DEFAULTrrWrmeanrinforr+PARQUET_ENCODING_RATIO_ESTIMATE_LOWER_BOUND)rrestimated_in_mem_size_arr file_size_arrestimated_encoding_ratiosestimated_ratios r<rrs  " $ $ =766 z??c)nn , , , ,!!LV!!!54)444M!!&)*C]&S&S!!! %766g788O KKMMMMMNNN  K L LLr>rrctt}g}|D]`}|||pt jjtg|||atdt|d}| |}| |S)N)scheduling_strategyretry_exceptions)r|r^zParquet dataset samplingfile)unit) rr|roptionsrrrOSErrorremoterrfetch_until_completeclose) rr|r^rfetch_file_infofuturesrL sample_barrs r<rrs''?@@OG%     # #$4%A*,,@") $   f 7WFSSSJ0099J r>target_block_sizecdSfd|D}|stSttjt j|d}t d|d|S)NcLg|] }||j|jdk|jz !Sr)ro)rrrs r<rz/_estimate_reader_batch_size..EsI  N'3'!++ B33 ,++r>r0z'Estimated parquet reader batch size at z rows)%DEFAULT_PARQUET_READER_ROW_BATCH_SIZErrrvrwrrrr)rravg_num_rows_per_blockestimated_batch_sizes ` r<rr?s t "544 #DIbg6L.M.M$N$NPQ R R KKU:NUUUVVV r>cddlm}t|dkr|d} |j|fi|d|i}n#t$r~ t |d\}}t j|tj j }|j|fi|d|i}n'#t$r}t||Yd}~nd}~wwxYwYn&t$r}t||Yd}~nd}~wwxYw|S)Nrr0rH)rHr) pyarrow.parquetrzrParquetDataset TypeErrorr(rrrrrrr") rrHr}pqdatasetresolved_pathsresolved_filesystemos_erXs r<rrYs       5zzQa(#"#     "    /// /2O$333 /N/#7";#!,!8!:!:!L### (b' /GG  / / / !$ . . . . . . . . / (((a''''''''( NsF6 C ABC  B8B3.C 3B88C = C CC csgStjttz}t t |t t}t |t}tj dtdz | t}fd| DS)Nrr0c g|] }| SrTrT)ridxrs r<rz%_sample_fragments..s 6 6 6sIcN 6 6 6r>) rvrwr.PARQUET_ENCODING_RATIO_ESTIMATE_SAMPLING_RATIOrrrs/PARQUET_ENCODING_RATIO_ESTIMATE_MAX_NUM_SAMPLES/PARQUET_ENCODING_RATIO_ESTIMATE_MIN_NUM_SAMPLESrlinspaceastyperQtolist)rtarget_num_samplespivotss` r<rr~s   IGG  OPP7 /Y@@[C NNQ.0B C C J J3 O OF 6 6 6 6fmmoo 6 6 66r>r\rBc,|D]~\}}|j|}|dkr)tj|||}Nt d|rtd|d|S)Nr%duplicate_partition_field_zThe partition field 'z`' also exists in the Parquet file. Ray Data will default to using the value in the Parquet file.) rir^r&rrLrMr,rr)r\rB partition_colrh field_indexs r<rNrNs!5 : : < < ul22=AA "  !+E22>>}eTTEE B=BB C C  NNW WWW    Lr> file_pathscddl}t|dkr|gS||gS|d}g}t|}||}|D]i}||jvr!||j|}n|}||||j||S)zReturn a new schema with partition fields added. This function infers the partition fields from the first file path in the dataset. rN) rVrr^r' field_typesfrom_numpy_dtypestringrr*) rrr_ first_pathr'rrjrg field_types r<rrs :!yy}}  yy}}AJ F  . .F ##J 88 1 1 1,,\-Ej-QRRJJJ  bhhz:667777 99V  r>user_specified_columnscfd|D}|.t|}|jfd|D}ng}||fS)aVInfer which columns are in the files and which columns are partition columns. This function uses the schema and path of the first file to infer what columns represent. Args: user_specified_columns: A list of column names that the user specified. fragment: The first fragment in the dataset. partitioning: The partitioning scheme used to partition the data. Returns: A tuple of lists of column names. The first list contains the columns that are in the file, and the second list contains the columns that are partition columns. c0g|]}|jjv|SrT)rVrX)rr+rLs r<rz5_infer_data_and_partition_columns..s4  X-3 3 3  3 3 3r>Ncg|]}|v| SrTrT)rr+rjs r<rz5_infer_data_and_partition_columns..s*   6Z;O;OF;O;O;Or>)r'rG)rrLrrrrirjs ` @r<rrs(,L #L11U8=))     !7    * **r>r8)lloggingrvrT dataclassesrtypingrrrrrr r r r r rnumpyrpackaging.versionr parse_versionrray._private.arrow_utilsrray.data._internal.arrow_blockrr>ray.data._internal.planner.plan_expression.expression_visitorsrray.data._internal.progress_barrray.data._internal.remote_fnrray.data._internal.utilrrrrray.data.blockrrrray.data.contextrray.data.datasourcerr^r)ray.data.datasource.file_based_datasourcer &ray.data.datasource.file_meta_providerr!r"r# ray.data.datasource.partitioningr$r%r&r'ray.data.datasource.path_utilr(ray.data.expressionsr)r*r+ray.util.debugr,rVpyarrow.datasetr- getLoggerrMrruNUM_CPUS_FOR_META_FETCH_TASKrFILE_READING_RETRYrrrrrrtr4rbrdrorrryr5rQrar6rr=rbr|rkrrrrrrrNrrrTr>r<rs !!!!!!                          444444 888888877777999999 ?>>>>>>>>>((((((******333333GGGGGG  =<<<<<<<<<######4NNN333333  8 $ $$1=#:#: #)/%+,'/0+26. 34/24/,0(55555555@0  ( ( ( ( ( ( (  (`P`P`P`P`P`P`PFkkkkk kkkr;?- - %01- CH~- #+3-- 49% - &d38n5 -  S * -  U4!556 7- $%- - - 67- o- - - - p;? $26eee#e e49% e &d38n5 e  S * ee67e eee S#X/eoeeeeP#S *,22d3i 2 % & 2  ! 2222j  B B B B B B B  B"M$%"M%&"M "M"M"M"MJ,-d3i  % &  tn   (# $% D X./0 EMc]  c]    4"""J7$%7 77772s$556?N"!!S !!!!!H"+ I"+#"+<("+ 49d3i  "+"+"+"+"+"+r>