&`iVddlZddlZddlZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZddlZddlmZmZe r ddlZddlmZeeeeeefZejeZ eGddeeZ!eeGd d Z"eGd d Z#ed GddZ$dedede fdZ%dS)N) dataclass)Enum) TYPE_CHECKINGAnyCallableDictListOptionalTypeUnion) DeveloperAPI PublicAPI)ExprceZdZdZdZdZdS)PartitionStyleaSupported dataset partition styles. Inherits from `str` to simplify plain text serialization/deserialization. Examples: >>> # Serialize to JSON text. >>> json.dumps(PartitionStyle.HIVE) # doctest: +SKIP '"hive"' >>> # Deserialize from JSON text. >>> PartitionStyle(json.loads('"hive"')) # doctest: +SKIP hivedirN)__name__ __module__ __qualname____doc__HIVE DIRECTORYt/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/datasource/partitioning.pyrrs$   DIIIrrceZdZUdZeed<dZeeed<dZ ee eed<dZ ee ee fed<dZeded<d Zed efd Zedd Zd ZdS) PartitioningaPartition scheme used to describe path-based partitions. Path-based partition formats embed all partition keys and values directly in their dataset file paths. For example, to read a dataset with `Hive-style partitions `_: >>> import ray >>> from ray.data.datasource.partitioning import Partitioning >>> ds = ray.data.read_csv( ... "s3://anonymous@ray-example-data/iris.csv", ... partitioning=Partitioning("hive"), ... ) Instead, if your files are arranged in a directory structure such as: .. code:: root/dog/dog_0.jpeg root/dog/dog_1.jpeg ... root/cat/cat_0.jpeg root/cat/cat_1.jpeg ... Then you can use directory-based partitioning: >>> import ray >>> from ray.data.datasource.partitioning import Partitioning >>> root = "s3://anonymous@air-example-data/cifar-10/images" >>> partitioning = Partitioning("dir", field_names=["class"], base_dir=root) >>> ds = ray.data.read_images(root, partitioning=partitioning) styleNbase_dir field_names field_typespyarrow.fs.FileSystem filesystemcZ|jd|_|ji|_d|_d|_dS)N)r r"_normalized_base_dir_resolved_filesystemselfs r __post_init__zPartitioning.__post_init__ns8 = DM   #!D $(!$(!!!rreturncF|j||jS)zJReturns the base directory normalized for compatibility with a filesystem.)r'_normalize_base_dirr)s rnormalized_base_dirz Partitioning.normalized_base_dirx(  $ ,  $ $ & & &((rcF|j||jS)zHReturns the filesystem resolved for compatibility with a base directory.)r(r.r)s rresolved_filesystemz Partitioning.resolved_filesystemr0rcddlm}||j|j\}|_t |dksJdt ||d}t |r|ds|dz }||_dS)aNormalizes the partition base directory for compatibility with the given filesystem. This should be called once a filesystem has been resolved to ensure that this base directory is correctly discovered at the root of all partitioned file paths. r)_resolve_paths_and_filesystemz0Expected 1 normalized base directory, but found /N)ray.data.datasource.path_utilr4r r$r(lenendswithr')r*r4pathsr/s rr.z Partitioning._normalize_base_dirs POOOOO+H+H M O, , (t( JJ!OOO Jc%jj J J OO#Ah " # # ',?,H,H,M,M ' 3 & $7!!!r)r,r#)rrrrr__annotations__r r strr!r r"rPartitionDataTyper$r+propertyr/r2r.rrrrr3s""J  #Hhsm""" (,K$s)$+++;?K$s$5567>>>48J01888))))S)))X) )))X) 88888rrc\eZdZdZeejddddfdedeedee edee ee fdedd df d Z d e fd Zd ed e eeffdZd eddd efdZed e fdZd ed eefdZded e eeffdZded e eeffdZdS)PathPartitionParseraPartition parser for path-based partition formats. Path-based partition formats embed all partition keys and values directly in their dataset file paths. Two path partition formats are currently supported - `HIVE` and `DIRECTORY`. For `HIVE` Partitioning, all partition directories under the base directory will be discovered based on `{key1}={value1}/{key2}={value2}` naming conventions. Key/value pairs do not need to be presented in the same order across all paths. Directory names nested under the base directory that don't follow this naming condition will be considered unpartitioned. If a partition filter is defined, then it will be called with an empty input dictionary for each unpartitioned file. For `DIRECTORY` Partitioning, all directories under the base directory will be interpreted as partition values of the form `{value1}/{value2}`. An accompanying ordered list of partition field names must also be provided, where the order and length of all partition values must match the order and length of field names. Files stored directly in the base directory will be considered unpartitioned. If a partition filter is defined, then it will be called with an empty input dictionary for each unpartitioned file. For example, if the base directory is `"foo"`, then `"foo.csv"` and `"foo/bar.csv"` would be considered unpartitioned files but `"foo/bar/baz.csv"` would be associated with partition `"bar"`. If the base directory is undefined, then `"foo.csv"` would be unpartitioned, `"foo/bar.csv"` would be associated with partition `"foo"`, and "foo/bar/baz.csv" would be associated with partition `("foo", "bar")`. Nrr r!r"r$r#r,cFt|||||}t|S)aCreates a path-based partition parser using a flattened argument list. Args: style: The partition style - may be either HIVE or DIRECTORY. base_dir: "/"-delimited base directory to start searching for partitions (exclusive). File paths outside of this directory will be considered unpartitioned. Specify `None` or an empty string to search for partitions in all file path directories. field_names: The partition key names. Required for DIRECTORY partitioning. Optional for HIVE partitioning. When non-empty, the order and length of partition key field names must match the order and length of partition directories discovered. Partition key field names are not required to exist in the dataset schema. field_types: A dictionary that maps partition key names to their desired data type. If not provided, the data type default to string. filesystem: Filesystem that will be used for partition path file I/O. Returns: The new path-based partition parser. )rr@)rr r!r"r$schemes rofzPathPartitionParser.ofs'8eX{KTT"6***r partitioningcJ|j}|j}|tjkr|st dtj|jtj|ji}|||_ |j 't d|d| ||_ dS)aCreates a path-based partition parser. Args: partitioning: The path-based partition scheme. The parser starts searching for partitions from this scheme's base directory. File paths outside the base directory will be considered unpartitioned. If the base directory is `None` or an empty string then this will search for partitions in all file path directories. Field names are required for DIRECTORY partitioning, and optional for HIVE partitioning. When non-empty, the order and length of partition key field names must match the order and length of partition directories discovered. zDirectory partitioning requires a corresponding list of partition key field names. Please retry your request with one or more field names specified.NzUnsupported partition style: z. Supported styles: ) rr!rr ValueErrorr_parse_hive_path_parse_dir_pathget _parser_fnkeys_scheme)r*rDrr!parserss r__init__zPathPartitionParser.__init__s"". N, , ,[ ,1   !6  $d&: %> $ 8 > > @ @ J J E9 +Ju,=y I IJu  r predicaterc0ddl}ddlm}||}|sdS |d|D}||}||}t ||j|jtj frt|dSddl } t || j rt|jdSt|S#t$r!t d|dYdSwxYw) a3Evaluate a predicate expression against partition values from a path. This method enables partition pruning by evaluating predicates that reference partition columns against the partition values parsed from file paths. Args: path: File path to parse partition values from. predicate: Expression that references partition columns. Returns: True if the partition satisfies the predicate (should read the file), False if it doesn't (can skip the file for partition pruning). rN)NativeExpressionEvaluatorFci|] \}}||g Srr).0colvals r zGPathPartitionParser.evaluate_predicate_on_partition..0s EEESseEEErzUFailed to evaluate predicate on partition for path %s, conservatively including file.T)exc_info)pyarrow?ray.data._internal.planner.plan_expression.expression_evaluatorr[tablerRvisit isinstanceArray ChunkedArraynpndarrayboolpandasSeriesiloc Exceptionloggerdebug) r*rOrYpar[partition_valuespartition_table evaluatorresultpds revaluate_predicate_on_partitionz3PathPartitionParser.evaluate_predicate_on_partitionse        4:: 5  hhEE,<,B,B,D,DEEEO 21/BBI__Y//F&28R_bj"IJJ 'F1I&   &"),, ,FKN+++<<     LL1     44 sB C*(2C*C**'DDc|jS)z)Returns the partitioning for this parser.)rLr)s rrBzPathPartitionParser.schemeM |rc||jjsdS|t|jjd}t j|S)zTrims the normalized base directory and returns the directory path. Returns None if the path does not start with the normalized base directory. Simply returns the directory path if the base directory is undefined. N) startswithrLr/r8 posixpathdirnamer*rOs rrQz'PathPartitionParser._dir_path_trim_baseRsP t|?@@ 4C 899;;< &&&rrTc d|dD}|r d|Dng}d|D}|jj}|r|rt|t|kr3t dt|dt|d|dt |D]8\}}||d |kr!t d |d ||d 9t |S) zHive partition path parser. Returns a dictionary mapping partition keys to values given a hive-style partition path of the form "{key1}={value1}/{key2}={value2}/..." or an empty dictionary for unpartitioned files. cHg|]}||ddk| S)=r5)countr]ds r z8PathPartitionParser._parse_hive_path..ds1LLLa!L9J9J9J9J9Jrr6c8g|]}|dS)r)splitrs rrz8PathPartitionParser._parse_hive_path..es"///QAGGCLL///rcVg|]&\}}|tj|g'Sr)urllibparseunquote)r]keyvalues rrz8PathPartitionParser._parse_hive_path..is1RRR:3S&,..u556RRRr Expected  partition value(s) but found : .rzExpected partition key z but found )rrLr!r8rF enumeratedict)r*rTdirskv_pairsr!i field_names rrGz$PathPartitionParser._parse_hive_path]sUML8>>#..LLL37?//$////RSRRRRl.  8 8}}K 0 000 4K 0 0448}}44(0444"+;!7!7   :A;q>Z//$,*,,#A;q>,,0 H~~rc >d|dD}|jj}|rSt|t|kr3t dt|dt|d|d|siSdt ||DS)afDirectory partition path parser. Returns a dictionary mapping directory partition keys to values from a partition path of the form "{value1}/{value2}/..." or an empty dictionary for unpartitioned files. Requires a corresponding ordered list of partition key field names to map the correct key to each value. cg|]}||Srrrs rrz7PathPartitionParser._parse_dir_path..s444a!4444rr6rrrrci|] \}}||| SNr)r]rV directorys rr`z7PathPartitionParser._parse_dir_path..s.    y  9   r)rrLr!r8rFzip)r*rTrr!s rrHz#PathPartitionParser._parse_dir_pathzs548>>#..444l.  CII[!1!111(C ,,((t99(( $(((   I  $' T$:$:    r)rrrr staticmethodrrr r<r rr=rCrrNrXrkrxr>rBrQrGrHrrrr@r@s: . 3"&+/>B8< +++3-+d3i(+d3(9#9:; + 45 +  +++\+<$\$$$$BST#s(^(;C;F;t;;;;z X ' ' ' ' ' 'c3h:  S#X      rr@beta) stabilitycPeZdZdZeejddddfdeee e fge fdede e de e e de ee e fde d d dfd Zd edeee e fge ffd Zde e d e e fdZde d e fdZed efdZdS)PathPartitionFilterzPartition filter for path-based partition formats. Used to explicitly keep or reject files based on a custom filter function that takes partition keys and values parsed from the file's path as input. N filter_fnrr r!r"r$r#r,cft|||||}t|}t||S)aCreates a path-based partition filter using a flattened argument list. Args: filter_fn: Callback used to filter partitions. Takes a dictionary mapping partition keys to values as input. Unpartitioned files are denoted with an empty input dictionary. Returns `True` to read a file for that partition or `False` to skip it. Partition keys and values are always strings read from the filesystem path. For example, this removes all unpartitioned files: .. code:: python lambda d: True if d else False This raises an assertion error for any unpartitioned file found: .. code:: python def do_assert(val, msg): assert val, msg lambda d: do_assert(d, "Expected all files to be partitioned!") And this only reads files from January, 2022 partitions: .. code:: python lambda d: d["month"] == "January" and d["year"] == "2022" style: The partition style - may be either HIVE or DIRECTORY. base_dir: "/"-delimited base directory to start searching for partitions (exclusive). File paths outside of this directory will be considered unpartitioned. Specify `None` or an empty string to search for partitions in all file path directories. field_names: The partition key names. Required for DIRECTORY partitioning. Optional for HIVE partitioning. When non-empty, the order and length of partition key field names must match the order and length of partition directories discovered. Partition key field names are not required to exist in the dataset schema. field_types: A dictionary that maps partition key names to their desired data type. If not provided, the data type defaults to string. filesystem: Filesystem that will be used for partition path file I/O. Returns: The new path-based partition filter. )rr@r)rrr r!r"r$rBpath_partition_parsers rrCzPathPartitionFilter.ofs9neX{KTT 3F ; ;"#8)DDDrrc"||_||_dS)aCreates a new path-based partition filter based on a parser. Args: path_partition_parser: The path-based partition parser. filter_fn: Callback used to filter partitions. Takes a dictionary mapping partition keys to values as input. Unpartitioned files are denoted with an empty input dictionary. Returns `True` to read a file for that partition or `False` to skip it. Partition keys and values are always strings read from the filesystem path. For example, this removes all unpartitioned files: ``lambda d: True if d else False`` This raises an assertion error for any unpartitioned file found: ``lambda d: assert d, "Expected all files to be partitioned!"`` And this only reads files from January, 2022 partitions: ``lambda d: d["month"] == "January" and d["year"] == "2022"`` N)_parser _filter_fn)r*rrs rrNzPathPartitionFilter.__init__s*- #rr:c6|}jfd|D}|S)a}Returns all paths that pass this partition scheme's partition filter. If no partition filter is set, then returns all input paths. If a base directory is set, then only paths under this base directory will be parsed for partitions. All paths outside of this base directory will automatically be considered unpartitioned, and passed into the filter function as empty dictionaries. Also normalizes the partition base directory for compatibility with the given filesystem before applying the filter. Args: paths: Paths to pass through the partition filter function. All paths should be normalized for compatibility with the given filesystem. Returns: List of paths that pass the partition filter, or all paths if no partition filter is defined. Nc>g|]}||Sr)apply)r]rOr*s rrz0PathPartitionFilter.__call__..s*IIIt 48H8HIdIIIr)r)r*r:filtered_pathss` rrXzPathPartitionFilter.__call__s1( ? &IIIIuIIINrrOcR|||Sr)rrrs rrzPathPartitionFilter.apply s t||D11222rc|jS)z2Returns the path partition parser for this filter.)rr)s rparserzPathPartitionFilter.parser rzr)rrrrrrrrrr<rkr r r=rCr@rNrXrr>rrrrrrs !/ 3"&+/>B8< 8E8ET#s(^,d238E8E3-8Ed3i( 8E d3(9#9:; 8E 45 8E 8E8E8E\8Et$2$T#s(^,d23$$$$0d3iDI23#3$3333+XrrrrWr,c|turt|S|turt|S|tur|dkS|S)Ntrue)intfloatrklower)rrWs rrSrSsSC5zz e  U|| d  {{}}&& r)&loggingr} urllib.parser dataclassesrenumrtypingrrrrr r r r numpyriray.util.annotationsr rrbray.data.expressionsrrrr<rkr= getLoggerrrprrr@rrSrrrrs_!!!!!!                    88888888*NNN))))))sE3456  8 $ $S$& e8e8e8e8e8e8e8 e8Pu u u u u u u u p Vyyyyyyyyxs'8Sr