o $iT @sddlZddlmZddlmZmZmZmZmZm Z m Z m Z ddl Z ddlZddlmZddlmZmZmZmZmZmZmZmZmZmZddlmZerZddlm Z ddl!m"Z"m#Z#e$e%Z&ed d eGd d d Z'eGd ddZ(de)deefddZ*de)deefddZ+de)deefddZ,dee dee)geefffddZ-de)dddeefddZ.de)dddee dee)geeffdeefddZ/  d;d e d!d"e ee)de ee dee)geeffde(fd#d$Z0d%ed&ed'ej"dee)e eej"fffd(d)Z1d*ee)e2fd+ej d,ej d-eede3f d.d/Z4 0dK/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/stats.py_safe_convert_table:s&    z"DatasetSummary._safe_convert_tabledfreturncCs |j|jvr ||jStS)zSet the statistic column as index if it exists, else return empty DataFrame. Args: df: DataFrame to set index on Returns: DataFrame with statistic column as index, or empty DataFrame if column missing )STATISTIC_COLUMNr" set_indexr6r7)r8rAr>r>r?_set_statistic_indexYs z#DatasetSummary._set_statistic_indexcCsb|||j}|||j}|jr |jr tj|jgdS||}| |jj ddS)aConvert summary to a single pandas DataFrame. Combines statistics from both schema-matching and schema-changing tables. Note: Some PyArrow extension types (like TensorExtensionType) may fail to convert to pandas when all values in a column are None. In such cases, this method attempts to convert column-by-column, casting problematic columns to null type. Returns: DataFrame with all statistics, where rows are unique statistics from both tables r"Tdrop) rEr@rr emptyr6r7rC combine_first reset_index sort_values)r8 df_matching df_changingresultr>r>r?r)fs    zDatasetSummary.to_pandasr2cCs4||jjvrdS|||j|g}|j|didS)aExtract a column from a PyArrow table if it exists. Args: table: PyArrow table to extract from column: Column name to extract Returns: DataFrame with 'statistic' and 'value' columns, or None if column doesn't exist NvaluerF)r0r1r@rCrename)r8r#r2rAr>r>r?_extract_column_from_tables z)DatasetSummary._extract_column_from_tablecszfddjjfD}|stddtj|dd}dd}|jjd d d | jjdd }|S) zGet all statistics for a specific column, merging from both tables. Args: column: Column name to get statistics for Returns: DataFrame with all statistics for the column cs$g|]}|durqS)N)rR).0r#r2rAr8r>r? s z3DatasetSummary.get_column_stats..zColumn 'z' not found in summary tablesT) ignore_indexcSs"|}t|dkr|jdSdS)Nr)dropnar4iloc)seriesnon_nullr>r>r?first_non_nullsz7DatasetSummary.get_column_stats..first_non_nullF)sortrPrG) rr r+r6concatgroupbyrCapplyrKrL)r8r2dfscombinedr[rOr>rTr?get_column_statss"  zDatasetSummary.get_column_statsN)__name__ __module__ __qualname____doc__rCr,Table__annotations__rliststrr@r6r7rEr)rdictrRrbr>r>r>r?rs$       rc@s.eZdZUdZeeefed<eeed<dS)_DtypeAggregatorszContainer for columns and their aggregators. Attributes: column_to_dtype: Mapping from column name to dtype string representation aggregators: List of all aggregators to apply column_to_dtype aggregatorsN) rcrdrerfrrjrhrr r>r>r>r?rls rlr2rBc CsVt|ddt|ddt|ddt|ddt|dddt|dgdt|dt|ddgS) aGenerate default metrics for numerical columns. This function returns a list of aggregators that compute the following metrics: - count - mean - min - max - std - approximate_quantile (median) - missing_value_percentage - zero_percentage Args: column: The name of the numerical column to compute metrics for. Returns: A list of AggregateFnV2 instances that can be used with Dataset.aggregate() Fon ignore_nullsTr)rprqddofg?)rp quantilesrp)rrrrrr rrr2r>r>r?_numerical_aggregatorss       rvcCs*t|ddt|ddt|ddt|dgS)a|Generate default metrics for temporal columns. This function returns a list of aggregators that compute the following metrics: - count - min - max - missing_value_percentage Args: column: The name of the temporal column to compute metrics for. Returns: A list of AggregateFnV2 instances that can be used with Dataset.aggregate() FroTrt)rrrrrur>r>r?_temporal_aggregatorss   rwcCs t|ddt|dt|ddgS)aGenerate default metrics for all columns. This function returns a list of aggregators that compute the following metrics: - count - missing_value_percentage - approximate_top_k (top 10 most frequent values) Args: column: The name of the column to compute metrics for. Returns: A list of AggregateFnV2 instances that can be used with Dataset.aggregate() Frort )rpk)rrrrur>r>r?_basic_aggregatorss  rzcCsddlm}m}|t|t|t|t|t| t| t| t| t| t|t|t|t|jtiS)aGet default mapping from Ray Data DataType to aggregator factory functions. This function returns factory functions that create aggregators for specific columns. Returns: Dict mapping DataType or TypeCategory to factory functions that take a column name and return a list of aggregators for that column. Examples: >>> from ray.data.datatype import DataType >>> from ray.data.stats import _default_dtype_aggregators >>> mapping = _default_dtype_aggregators() >>> factory = mapping.get(DataType.int32()) >>> aggs = factory("my_column") # Creates aggregators for "my_column" rr)ray.data.datatyperrint8rvint16int32int64uint8uint16uint32uint64float32float64boolstringrzbinaryTEMPORALrwrr>r>r?_default_dtype_aggregatorss rdtyperc Csz*|rtj|jrt|ddgWS|rt|WS|r&t |WSt |WSt yO}zt d|d|d|dt |WYd}~Sd}~ww)a!Get aggregators using heuristic-based type detection. This is a fallback when no explicit mapping is found for the dtype. Args: column: Column name dtype: Ray Data DataType for the column Returns: List of aggregators suitable for the column type Froz,Could not determine aggregators for column 'z ' with dtype z: z. Using basic aggregators.N) is_arrow_typer,typesis_null_physical_dtyperis_numerical_typervis_temporal_typerwrz Exceptionr.r/)r2rr9r>r>r?_get_fallback_aggregatorsDs    rdtype_agg_mappingcCsnddlm}m}|D]%\}}t||r||kr||St||tfr1||r1||Sq t||S)aGet aggregators for a specific column based on its DataType. Attempts to match the dtype against the provided mapping first, then falls back to heuristic-based selection if no match is found. Args: column: Column name dtype: Ray Data DataType for the column dtype_agg_mapping: Mapping from DataType to factory functions Returns: List of aggregators with the column name properly set rr)r{rritems isinstancerjis_ofr)r2rrrr mapping_keyfactoryr>r>r?_get_aggregators_for_dtypeds   rr0rr"cCs ddlm}|s td|dur|j}t|t|j}|r&td|dt}|rA|}|D] \}}||vr?|||<q3n|}i} g} tt |j|j } |D],} | | } | dus`| t urjt d| dqR|| }t|| | <| t| ||qRt| | d S) aGenerate aggregators for columns in a dataset based on their DataTypes. Args: schema: A Ray Schema instance columns: List of columns to include. If None, all columns will be included. dtype_agg_mapping: Optional user-provided mapping from DataType to aggregator factories. Each value should be a callable that takes a column name and returns aggregators. This will be merged with the default mapping (user mapping takes precedence). Returns: _DtypeAggregators containing column-to-dtype mapping and aggregators Raises: ValueError: If schema is None or if specified columns don't exist in schema rrz4Dataset must have a schema to determine column typesNzColumns z not found in dataset schemazSkipping field 'z': type is None or unsupported)rmrn)r{rr+r1setrcopyrrkziprobjectr.r/ from_arrowrjextendrrl)r0r"rr missing_colsdefaults final_mappingryvrmall_aggs name_to_typenamepa_type ray_dtyper>r>r?_dtype_aggregators_for_datasets@   raggrPagg_typecsddlm}|tj|p||}t|t s"|durK|rK||r,|j n||dur3nddt t |D}fddt ||DS||fiS)aFormat aggregation result into stat entries. Takes the raw aggregation result and formats it into one or more stat entries. For scalar results, returns a single entry. For list results, expands into multiple indexed entries. Args: agg: The aggregator instance value: The aggregation result value agg_type: PyArrow type of the aggregation result Returns: Dictionary mapping stat names to (value, type) tuples rrNcSsg|]}t|qSr>)rj)rSidxr>r>r?rUz!_format_stats..cs&i|]\}}d|d|fqS)[]r>)rSlabellist_valagg_name scalar_typer>r? sz!_format_stats..)r{r get_agg_namer,ris_listr is_list_typerri value_typeranger4r)rrPrrrlabelsr>rr? _format_statss    r agg_resultoriginal_schema agg_schemarncCsi}i}t}dd|D}|D]Q\}} d|vs|ds q||} | s(q| } | s/q||j} || j} t| | | }|D]\}\}}|| krQ|n|}||f||i| <qE| | q|||fS)aParse aggregation results into schema-matching and schema-changing stats. Args: agg_result: Dictionary of aggregation results with keys like "count(col)" original_schema: Original dataset schema agg_schema: Schema of aggregation results aggregators: List of aggregators used to generate the results Returns: Tuple of (schema_matching_stats, schema_changing_stats, column_names) cSsi|]}|j|qSr>)r)rSrr>r>r?rrz(_parse_summary_stats..()) rrendswithgetget_target_columnfieldr&r setdefaultadd)rrrrnschema_matchingschema_changingr" agg_lookupkeyrPrr;r original_typeformatted_stats stat_name stat_value stat_type stats_dictr>r>r?_parse_summary_statss,      rcol_datacol_typer;c CsB|durztj||dWStjtjfyYnwt||pdS)a{Create a PyArrow array with fallback strategies. Uses convert_to_pyarrow_array from arrow_block.py for type inference and error handling when no specific type is provided. Args: col_data: List of column values col_type: Optional PyArrow type to use col_name: Column name for error messages (optional) Returns: PyArrow array Nr%r2)r,arrayArrowTypeErrorr-r )rrr;r>r>r?_create_pyarrow_array/srr all_columnspreserve_typesc Cs|stiSt|}tj|i}t|D]B}g}d}|D]!} ||| vr:|| |\} } || |dur9| }q|dq|rN||jvrN||j } n|} t || |||<qt|S)asBuild a PyArrow table from parsed statistics. Args: stats_dict: Nested dict of {stat_name: {col_name: (value, type)}} all_columns: Set of all column names across both tables original_schema: Original dataset schema preserve_types: If True, use original schema types for columns Returns: PyArrow table with statistics N) r,r#sortedkeysrrCappendr1rr&r) rrrr stat_names table_datar;r first_typerrPrrr>r>r?_build_summary_tableKs(        r)NN)Nr);logging dataclassesrtypingrrrrrrr r pandasr6pyarrowr,$ray.air.util.tensor_extensions.arrowr ray.data.aggregater r rrrrrrrrray.util.annotationsrray.data.datasetrr{rr getLoggerrcr.rrlrjrvrwrzrrrrranytuplerArrayrrrrgrr>r>r>r?s ( 0      ,  #  C 0  8