Repository: ray-project/ray
File: python/ray/data/tests/test_stats.py (Python)

Code before:

    # Split the input string into lines and iterate over them
    for line in metrics_str.split("\n"):
        match = pattern.match(line)
        if match:
            # Extracting the operator name, time, and percentage
            operator_name, time_str, percent_str = match.groups()
            # Converting percentage to float and keeping time as string
            metrics_dict[operator_name] = (
                time_to_seconds(time_str),
                float(percent_str),
            )

    total_time, total_percent = metrics_dict.pop("Total")
    assert total_percent == 100

    # Tolerance for floating-point rounding errors (100ms)
    # Individual operator times may appear slightly larger than total time
    # due to rounding (e.g., 2.265s rounds to 2.27s for operator but 2.26s for total)
    TOLERANCE = 0.02

    for name, (time_s, percent) in metrics_dict.items():
        if name == "Scheduling":
            continue
        if time_s > total_time + TOLERANCE:
            print("runtime_metrics raw:\n", metrics_str)
            print("runtime_metrics parsed:", metrics_dict)
            print(
                f"runtime_metrics mismatch for '{name}': {time_s}s > {total_time}s (tolerance: {TOLERANCE}s)"
            )

        assert time_s <= total_time + TOLERANCE
        # Check percentage, this is done with some expected loss of precision
        # due to rounding in the intital output.
        assert isclose(percent, time_s / total_time * 100, rel_tol=0.01)


def test_per_node_metrics_basic(ray_start_regular_shared, restore_data_context):
    """Basic test to ensure per-node metrics are populated."""
    ctx = DataContext.get_current()
    ctx.enable_per_node_metrics = True

    def _sum_net_metrics(per_node_metrics: Dict[str, NodeMetrics]) -> Dict[str, float]:
        sum_metrics = defaultdict(float)
        for metrics in per_node_metrics.values():
            for metric, value in metrics.items():
                sum_metrics[metric] += value
        return sum_metrics

    with patch("ray.data._internal.stats.get_or_create_stats_actor") as mock_get_actor:
        mock_actor_handle = MagicMock()

Review comment (suggestion):
Can you add a comment describing we special-case `"Scheduling"`? I don't think it'll be completely obvious to readers 

Code after:

    # Split the input string into lines and iterate over them
    for line in metrics_str.split("\n"):
        match = pattern.match(line)
        if match:
            # Extracting the operator name, time, and percentage
            operator_name, time_str, percent_str = match.groups()
            # Converting percentage to float and keeping time as string
            metrics_dict[operator_name] = (
                time_to_seconds(time_str),
                float(percent_str),
            )

    total_time, total_percent = metrics_dict.pop("Total")
    assert total_percent == 100

    # Tolerance for floating-point rounding errors (100ms)
    # Individual operator times may appear slightly larger than total time
    # due to rounding (e.g., 2.265s rounds to 2.27s for operator but 2.26s for total)
    TOLERANCE = 0.02

    for name, (time_s, percent) in metrics_dict.items():
        # Special-case Scheduling: it's cumulative time across scheduling loops,
        # so it can exceed the wall-clock Total span under concurrency.
        if name == "Scheduling":
            continue
        if time_s > total_time + TOLERANCE:
            print("runtime_metrics raw:\n", metrics_str)
            print("runtime_metrics parsed:", metrics_dict)
            print(
                f"runtime_metrics mismatch for '{name}': {time_s}s > {total_time}s (tolerance: {TOLERANCE}s)"
            )

        assert time_s <= total_time + TOLERANCE
        # Check percentage, this is done with some expected loss of precision
        # due to rounding in the intital output.
        assert isclose(percent, time_s / total_time * 100, rel_tol=0.01)


def test_per_node_metrics_basic(ray_start_regular_shared, restore_data_context):
    """Basic test to ensure per-node metrics are populated."""
    ctx = DataContext.get_current()
    ctx.enable_per_node_metrics = True

    def _sum_net_metrics(per_node_metrics: Dict[str, NodeMetrics]) -> Dict[str, float]:
        sum_metrics = defaultdict(float)
        for metrics in per_node_metrics.values():
            for metric, value in metrics.items():
                sum_metrics[metric] += value
        return sum_metrics