Function _format_stats

python/ray/data/stats.py:456–501

Format 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:

(
    agg: AggregateFnV2, value: Any, agg_type: pa.DataType
)

def _format_stats(
    agg: AggregateFnV2, value: Any, agg_type: pa.DataType
) -> Dict[str, Tuple[Any, pa.DataType]]:
    """Format 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
    """
    from ray.data.datatype import DataType

    agg_name = agg.get_agg_name()

    # Handle list results: expand into separate indexed stats
    # If the value is None but the type is list, it means we got a null result
    # for a list-type aggregator (e.g., ignore_nulls=True and all nulls).
    is_list_type = (
        pa.types.is_list(agg_type) or DataType.from_arrow(agg_type).is_list_type()
    )

    if isinstance(value, list) or (value is None and is_list_type):
        scalar_type = (
            agg_type.value_type
            if DataType.from_arrow(agg_type).is_list_type()
            else agg_type
        )
        if value is None:
            # Can't expand None without knowing the size, return as-is
            pass
        else:
            labels = [str(idx) for idx in range(len(value))]
            return {
                f"{agg_name}[{label}]": (list_val, scalar_type)
                for label, list_val in zip(labels, value)
            }

    # Fallback to scalar result for non-list values or unexpandable Nones
    return {agg_name: (value, agg_type)}


def _parse_summary_stats(