&`iE ddlZddlmZmZmZmZmZmZddlZ ddl Z ddl m Z ddlmZmZmZddlmZddlmZddlmZmZmZGdd ZGd d ZeGd d eeZeGddZGddeZeGddegeefZeGddeZ dS)N)CallableDict GeneratorIterableListOptional)_check_pyarrow_version)Block BlockMetadataSchema)_iter_sliced_blocks)Expr) Deprecated DeveloperAPI PublicAPIc eZdZdZdefdZdeeeeffdZ deeeeffdZ dee efdZ e deeeefdeeeefdeeeeffd Zd eeeefdd fd Ze d ddeeeefddfdZe deddeeeefdedfdZdS)"_DatasourceProjectionPushdownMixinz:Mixin for reading operators supporting projection pushdownreturncdS)zvReturns ``True`` in case ``Datasource`` supports projection operation being pushed down into the reading layerFselfs r/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/datasource/datasource.pysupports_projection_pushdownz?_DatasourceProjectionPushdownMixin.supports_projection_pushdowns uc|jS)aReturn the projection map (original column names -> final column names). Returns: Dict mapping original column names (in storage) to final column names (after optional renames). Keys indicate which columns are selected. None means all columns are selected with no renames. Empty dict {} means no columns are selected. )_projection_maprs rget_projection_mapz5_DatasourceProjectionPushdownMixin.get_projection_maps ##rcf|jdSd|jD}|r|ndS)aNReturn the column renames from the projection map. This is used by predicate pushdown to rewrite filter expressions from renamed column names back to original column names. Returns: Dict mapping original column names to renamed names, or None if no renaming has been applied. Nc&i|]\}}||k ||Srr).0kvs r zI_DatasourceProjectionPushdownMixin.get_column_renames...s#KKKDAqAFF1aFFFr)ritems)rrenamess rget_column_renamesz5_DatasourceProjectionPushdownMixin.get_column_renames!sC   '4KKD$8$>$>$@$@KKK!+wwt+rc`|j&t|jndS)aQExtract data columns from projection map. Helper method for datasources that need to pass columns to legacy read functions expecting separate columns and rename_map parameters. Returns: List of column names, or None if all columns should be read. Empty list [] means no columns. N)rlistkeysrs r_get_data_columnsz4_DatasourceProjectionPushdownMixin._get_data_columns1s4#/ %**,, - - - rprev_projection_mapnew_projection_mapcp||S||Si}|D]\}}||vr ||}|||<|S)aCombine two projection maps via transitive composition. Args: prev_projection_map: Previous projection (original -> intermediate names) new_projection_map: New projection to apply (intermediate -> final names) Returns: Combined projection map (original -> final names) Examples: >>> # Select columns a, b with no renames >>> prev = {"a": "a", "b": "b"} >>> # Select only 'a', rename to 'x' >>> new = {"a": "x"} >>> _DatasourceProjectionPushdownMixin._combine_projection_map(prev, new) {'a': 'x'} >>> # First rename a->temp >>> prev = {"a": "temp"} >>> # Then rename temp->final >>> new = {"temp": "final"} >>> _DatasourceProjectionPushdownMixin._combine_projection_map(prev, new) {'a': 'final'} )r%)r,r-composedorig_colintermediate_name final_names r_combine_projection_mapz:_DatasourceProjectionPushdownMixin._combine_projection_mapAsm<  &% %  '& &+>+D+D+F+F 0 0 'H' $666/0AB %/" rprojection_map Datasourcecntj|}||j||_|S)aApply a projection to this datasource. Args: projection_map: Dict mapping original column names (in storage) to final column names (after optional renames). Keys indicate which columns to select. None means select all columns with no renames. Returns: A new datasource instance with the projection applied. )copyr3r)rr4clones rapply_projectionz3_DatasourceProjectionPushdownMixin.apply_projectionss: $!% < <  .! !  rtablepa.Tablecolumn_rename_mapcfs|Sfd|jjD}||S)zApply column renaming to a PyArrow table. Args: table: PyArrow table to rename column_rename_map: Mapping from old column names to new names Returns: Table with renamed columns c<g|]}||Sr)get)r!colr<s r zD_DatasourceProjectionPushdownMixin._apply_rename..s*SSS&**344SSSr)schemanamesrename_columns)r:r< new_namess ` r _apply_renamez0_DatasourceProjectionPushdownMixin._apply_renamesD! LSSSS @RSSS ##I...rtables)r;NNc#NK|D]}t||V dS)aWrap a table generator to apply column renaming to each table. This helper eliminates duplication across datasources that need to apply column renames to tables yielded from generators. Args: tables: Iterator/generator yielding PyArrow tables column_rename_map: Mapping from old column names to new names Yields: pa.Table: Tables with renamed columns N)rrF)rGr<r:s r_apply_rename_to_tablesz:_DatasourceProjectionPushdownMixin._apply_rename_to_tablessN"  E4BB(      rN)__name__ __module__ __qualname____doc__boolrrrstrrr'rr+ staticmethodr3r9rFrrrIrrrrrsDDd $HT#s(^$< $ $ $ $,HT#s(^$<,,,,  8DI#6    /%d38n5/$T#s(^4/ $sCx. !///\/b c3h0 .//#DcN3/ ///\/&$#DcN3 ) *\rrcLeZdZdZdZdefdZdeefdZ deddfdZ d S) !_DatasourcePredicatePushdownMixinz9Mixin for reading operators supporting predicate pushdowncd|_dSN_predicate_exprrs r__init__z*_DatasourcePredicatePushdownMixin.__init__s/3rrcdS)NFrrs rsupports_predicate_pushdownz=_DatasourcePredicatePushdownMixin.supports_predicate_pushdownsurc|jSrTrUrs rget_current_predicatez7_DatasourcePredicatePushdownMixin.get_current_predicates ##rpredicate_exprr5c^ddl}|j|}|j|n |j|z|_|S)aApply a predicate to this datasource. Default implementation that combines predicates using AND. Subclasses that support predicate pushdown should have a _predicate_expr attribute to store the predicate. Note: Column rebinding is handled by the PredicatePushdown rule before this method is called, so the predicate_expr should already reference the correct column names. rN)r7rV)rr\r7r8s rapply_predicatez1_DatasourcePredicatePushdownMixin.apply_predicatesH   $ $, N&7   rN) rJrKrLrMrWrNrYrrr[r^rrrrRrRsCC444T$x~$$$$ rrRceZdZdZdZeddZedededfdZ de fd Z de efd Z dded e ededfd ZedefdZedefdZd S)r5zInterface for defining a custom :class:`~ray.data.Dataset` datasource. To read a datasource into a dataset, use :meth:`~ray.data.read_datasource`. c:t|dS)z)Initialize the datasource and its mixins.N)rRrWrs rrWzDatasource.__init__s)22488888rrReaderc t|fi|Sz Deprecated: Implement :meth:`~ray.data.Datasource.get_read_tasks` and :meth:`~ray.data.Datasource.estimate_inmemory_data_size` instead. )_LegacyDatasourceReader)r read_argss r create_readerzDatasource.create_readers 't99y999r parallelismReadTaskc trcNotImplementedError)rrgres r prepare_readzDatasource.prepare_reads "!rct|j}d}||r|dt| }|S)ztReturn a human-readable name for this datasource. This will be used as the names of the read tasks. r5N)typerJendswithlen)rnamedatasource_suffixs rget_namezDatasource.get_namesNDzz"( ==* + + 313011112D rctzReturn an estimate of the in-memory data size, or None if unknown. Note that the in-memory data size may be larger than the on-disk data size. rjrs restimate_inmemory_data_sizez&Datasource.estimate_inmemory_data_size "!rNper_task_row_limitct)aExecute the read and return read tasks. Args: parallelism: The requested read parallelism. The number of read tasks should equal to this value if possible. per_task_row_limit: The per-task row limit for the read tasks. Returns: A list of read tasks that can be executed to read blocks from the datasource in parallel. rjrrgrxs rget_read_taskszDatasource.get_read_tasks s "!rct|jtju}t|jtju}| p| SrT)rnr{r5rv)rhas_implemented_get_read_tasks+has_implemented_estimate_inmemory_data_sizes rshould_create_readerzDatasource.should_create_readersS JJ %Z-F F ' JJ 29 : 4 / . ?>> rcdS)z:If ``False``, only launch read tasks on the driver's node.Trrs rsupports_distributed_readsz%Datasource.supports_distributed_reads&s tr)rrarT)rJrKrLrMrWrrfintrrlrOrsrrvr{propertyrNrrrrrr5r5sI 999:::Z:""T*=M"""Z"#"Xc]""""EI " " "4` to know the in-memory size of the read prior to executing it. rctrurjrs rrvz"Reader.estimate_inmemory_data_size5rwrrgrhct)aMExecute the read and return read tasks. Args: parallelism: The requested read parallelism. The number of read tasks should equal to this value if possible. Returns: A list of read tasks that can be executed to read blocks from the datasource in parallel. rj)rrgs rr{zReader.get_read_tasks<s "!rN) rJrKrLrMrrrvrr{rrrrara,sd"Xc]"""" "# "$z2B " " " " " "rracbeZdZdefdZdeefdZ d dedeededfd Z dS) rd datasourcec "||_||_dSrT) _datasource _read_args)rrres rrWz _LegacyDatasourceReader.__init__Ks%#rrcdSrTrrs rrvz3_LegacyDatasourceReader.estimate_inmemory_data_sizeOstrNrgrxrhc2|jj|fi|jS)aExecute the read and return read tasks. Args: parallelism: The requested read parallelism. The number of read tasks should equal to this value if possible. per_task_row_limit: The per-task row limit for the read tasks. Returns: A list of read tasks that can be executed to read blocks from the datasource in parallel. )rrlrrzs rr{z&_LegacyDatasourceReader.get_read_tasksRs$-t,[LLDOLLLrrT) rJrKrLr5rWrrrvrr{rrrrdrdJs$:$$$$Xc]EIMMM4||_||_||_||_dSrT) _metadata_read_fn_schema_per_task_row_limit)rrrrBrxs rrWzReadTask.__init__ys'"  #5   rrc|jSrT)rrs rrzReadTask.metadatas ~rc|jSrT)rrs rrBzReadTask.schemas |rc|jSrT)rrs rrzReadTask.read_fns }rc|jS)z.Get the per-task row limit for this read task.)rrs rrxzReadTask.per_task_row_limits ''rc#K|}t|ds"td||j |Ed{VdSt ||jEd{VdS)N__iter__zlRead function must return Iterable[Block], got {}. Probably you need to return `[block]` instead of `block`.)rhasattrDeprecationWarningformatrr )rresults r__call__zReadTask.__call__svz**  !6&>>     # +        F&vt/GHHHHHHHHHHHr)NN)rJrKrLrMrrr r rrrWrrrBrrxrrrrrhrhcsU0&*,0 6 6"huo-. 6  6" 6 %SM 6 6 6 6-X*X"huo"56X(HSM(((X( I(5/ I I I I I IrrhcveZdZdZdedefdZdeefdZ d ded eedee fd Z de fd Z dS) RandomIntRowDatasourceaAn example datasource that generates rows with random int64 columns. Examples: >>> import ray >>> from ray.data.datasource import RandomIntRowDatasource >>> source = RandomIntRowDatasource() # doctest: +SKIP >>> ray.data.read_datasource( # doctest: +SKIP ... source, n=10, num_columns=2).take() {'c_0': 1717767200176864416, 'c_1': 999657309586757214} {'c_0': 4983608804013926748, 'c_1': 1160140066899844087} n num_columnsc"||_||_dS)zInitialize the datasource that generates random-integer rows. Args: n: The number of rows to generate. num_columns: The number of columns to generate. N_n _num_columns)rrrs rrWzRandomIntRowDatasource.__init__s'rrc&|j|jzdzS)Nrrs rrvz2RandomIntRowDatasource.estimate_inmemory_data_sizesw**Q..rNrgrxc tddl g}|j}|j}t d||z}dt dt dt f fd jdt|Dj }d}||krct|||z } t| d| z|zdd } | t| |f fd | || ||z }||kc|S) Nrcountrrcjtjtjtjj||ftjdt|DS)N)sizedtypecg|]}d|S)c_rr!is rrAzMRandomIntRowDatasource.get_read_tasks..make_block..s<<.make_blockst=,, !!HRX&&*+u1ERX"=<{););<<< - rci|] }d|dg S)rrrrs rr$z9RandomIntRowDatasource.get_read_tasks..s" 7 7 7qX!XXs 7 7 7rr)num_rows size_bytes input_files exec_statsc||gSrTr)rrrs rz7RandomIntRowDatasource.get_read_tasks..s" 5+66Br)rBrx)r rrrrrr r from_pydictrrBminr appendrh) rrgrx read_tasksrr block_sizerBrrmetarrs @@rr{z%RandomIntRowDatasource.get_read_taskssm    %' G' A,--  c        ** 7 7E+$6$6 7 7 7    !ee AE**E u9{2  D   !&K!'9     OA%!ee(rcdS)zReturn a human-readable name for this datasource. This will be used as the names of the read tasks. Note: overrides the base `Datasource` method. RandomIntrrs rrszRandomIntRowDatasource.get_names {rrT) rJrKrLrMrrWrrvrrhr{rOrsrrrrrs  (#(C((((/Xc]//// -1...%SM. h ....`#rr)!r7typingrrrrrrnumpyrrparay.data._internal.utilr ray.data.blockr r r ray.data.datasource.utilr ray.data.expressionsrray.util.annotationsrrrrrRr5rardrhrrrrrsb FFFFFFFFFFFFFFFF::::::7777777777888888%%%%%%DDDDDDDDDDddddddddN%%%%%%%%P KKKKK35VKK K\ """"""" ":MMMMMfMMM2?I?I?I?I?IxHUO+,?I?I?IDOOOOOZOOOOOr