code: def _row_groups_to_parts(
      gather_statistics,
      split_row_groups,
      aggregation_depth,
      file_row_groups,
      file_row_group_stats,
      file_row_group_column_stats,
      stat_col_indices,
      make_part_func,
      make_part_kwargs,
  ):
      # Construct `parts` and `stats`
      parts = []
      stats = []
      if split_row_groups:
          # Create parts from each file,
          # limiting the number of row_groups in each piece
          split_row_groups = int(split_row_groups)
          residual = 0
          for filename, row_groups in file_row_groups.items():
              row_group_count = len(row_groups)
              if residual:
                  _rgs = [0] + list(
                      range(residual, row_group_count, split_row_groups)
                  )
              else:
                  _rgs = list(
                      range(residual, row_group_count, split_row_groups)
                  )
  
              for i in _rgs:
                  i_end = i + split_row_groups
                  if aggregation_depth is True:
                      if residual and i == 0:
                          i_end = residual
                          residual = 0
                      _residual = i_end - row_group_count
                      if _residual > 0:
                          residual = _residual
  
                  rg_list = row_groups[i:i_end]
  
                  part = make_part_func(
                      filename,
                      rg_list,
                      **make_part_kwargs,
                  )
                  if part is None:
                      continue
  
                  parts.append(part)
                  if gather_statistics:
                      stat = _aggregate_stats(
                          filename,
                          file_row_group_stats[filename][i:i_end],
                          file_row_group_column_stats[filename][
                              i:i_end
                          ],
                          stat_col_indices,
                      )
                      stats.append(stat)
      else:
          for filename, row_groups in file_row_groups.items():
              part = make_part_func(
                  filename,
                  row_groups,
                  **make_part_kwargs,
              )
              if part is None:
                  continue
  
              parts.append(part)
              if gather_statistics:
                  stat = _aggregate_stats(
                      filename,
                      file_row_group_stats[filename],
                      file_row_group_column_stats[filename],
                      stat_col_indices,
                  )
                  stats.append(stat)
  
      return parts, stats
  
imports: from __future__ import annotations
  import re
  import warnings
  import numpy
  import pandas
  from fsspec.core import expand_paths_if_needed, get_fs_token_paths, stringify_path
  from fsspec.spec import AbstractFileSystem
  from dask import config
  from dask.dataframe.io.utils import _is_local_fs
  from dask.utils import natural_sort_key, parse_bytes
label_desc: Use image node_id 6 for calling a global function with example usage: _row_groups_to_parts(gather_statistics, split_row_groups, aggregation_depth, file_row_groups, file_row_group_stats, file_row_group_column_stats, stat_col_indices, make_part_func, make_part_kwargs) and returns: parts,  stats