code: def merge_quantiles(finalq, qs, vals):
      """Combine several quantile calculations of different data.
      [NOTE: Same logic as dask.array merge_percentiles]
      """
      if isinstance(finalq, Iterator):
          finalq = list(finalq)
      finalq = np.array(finalq)
      qs = list(map(list, qs))
      vals = list(vals)
      vals, Ns = zip(*vals)
      Ns = list(Ns)
  
      L = list(
          zip(*[(q, val, N) for q, val, N in zip(qs, vals, Ns) if N])
      )
      if not L:
          raise ValueError("No non-trivial arrays found")
      qs, vals, Ns = L
  
      if len(vals) != len(qs) or len(Ns) != len(qs):
          raise ValueError(
              "qs, vals, and Ns parameters must be the same length"
          )
  
      # transform qs and Ns into number of observations between quantiles
      counts = []
      for q, N in zip(qs, Ns):
          count = np.empty(len(q))
          count[1:] = np.diff(q)
          count[0] = q[0]
          count *= N
          counts.append(count)
  
      def _append_counts(val, count):
          val["_counts"] = count
          return val
  
      # Sort by calculated quantile values, then number of observations.
      combined_vals_counts = gd.core.reshape._merge_sorted(
          [*map(_append_counts, vals, counts)]
      )
      combined_counts = cupy.asnumpy(
          combined_vals_counts["_counts"].values
      )
      combined_vals = combined_vals_counts.drop(columns=["_counts"])
  
      # quantile-like, but scaled by total number of observations
      combined_q = np.cumsum(combined_counts)
  
      # rescale finalq quantiles to match combined_q
      desired_q = finalq * sum(Ns)
  
      # TODO: Support other interpolation methods
      # For now - Always use "nearest" for interpolation
      left = np.searchsorted(combined_q, desired_q, side="left")
      right = np.searchsorted(combined_q, desired_q, side="right") - 1
      np.minimum(
          left, len(combined_vals) - 1, left
      )  # don't exceed max index
      lower = np.minimum(left, right)
      upper = np.maximum(left, right)
      lower_residual = np.abs(combined_q[lower] - desired_q)
      upper_residual = np.abs(combined_q[upper] - desired_q)
      mask = lower_residual > upper_residual
      index = lower  # alias; we no longer need lower
      index[mask] = upper[mask]
      rv = combined_vals.iloc[index]
      return rv.reset_index(drop=True)
  
imports: import warnings
  from collections.abc import Iterator
  from functools import wraps
  import cupy
  import numpy
  import tlz
  from dask import config
  from dask.base import tokenize
  from dask.dataframe import methods
  from dask.dataframe.core import DataFrame, Index, Series
  from dask.dataframe.shuffle import rearrange_by_column
  from dask.highlevelgraph import HighLevelGraph
  from dask.utils import M
  import cudf
  from cudf.api.types import is_categorical_dtype
  from cudf.utils.nvtx_annotation import _dask_cudf_nvtx_annotate
label_desc: Use image node_id 5 for calling a global function with example usage: merge_quantiles(finalq, qs, vals) and returns: rv, val