code: def _null_wrap_accumulate_row(
      ignore_nulls: bool,
      on_fn: Callable[[T], T],
      accum: Callable[[AggType, T], AggType],
  ) -> Callable[[WrappedAggType, T], WrappedAggType]:
      """
      Wrap accumulator function with null handling.
  
      The returned accumulate function expects a to be either None or of the form:
      a = [acc_data_1, ..., acc_data_n, has_data].
  
      This performs an accumulation subject to the following null rules:
      1. If r is null and ignore_nulls=False, return None.
      2. If r is null and ignore_nulls=True, return a.
      3. If r is non-null and a is None, return None.
      4. If r is non-null and a is non-None, return accum(a[:-1], r).
  
      Args:
          ignore_nulls: Whether nulls should be ignored or cause a None result.
          on_fn: Function selecting a subset of the row to apply the aggregation.
          accum: The core accumulator function to wrap.
  
      Returns:
          A new accumulator function that handles nulls.
      """
  
      def _accum(a: WrappedAggType, r: T) -> WrappedAggType:
          r = on_fn(r)
          if _is_null(r):
              if ignore_nulls:
                  # Ignoring nulls, return the current accumulation, ignoring r.
                  return a
              else:
                  # Not ignoring nulls, so propagate the null.
                  return None
          else:
              if a is None:
                  # Accumulation is None so (1) a previous row must have been null, and
                  # (2) we must be propagating nulls, so continue to pragate this null.
                  return None
              else:
                  # Row is non-null and accumulation is non-null, so we now apply the core
                  # accumulation.
                  a, _ = _unwrap_acc(a)
                  a = accum(a, r)
                  return _wrap_acc(a, has_data=True)
  
      return _accum
  
file_path: ray/python/ray/data/_internal/null_aggregate.py
imports: from types import ModuleType
  from typing import Any, Callable, Tuple, Union
  import numpy
  from ray.data.block import AggType, Block, KeyType, T, U
label_desc: Use image node_id 5 for calling a global function with example usage: _null_wrap_accumulate_row(ignore_nulls, on_fn, accum) and returns: _accum, a, None, None, _wrap_acc