code: def new_combiner(
      *,
      output_path: str,
      temp_path: str,
      save_path: Optional[str] = None,
      gvcf_paths: Optional[List[str]] = None,
      vds_paths: Optional[List[str]] = None,
      vds_sample_counts: Optional[List[int]] = None,
      intervals: Optional[List[Interval]] = None,
      import_interval_size: Optional[int] = None,
      use_genome_default_intervals: bool = False,
      use_exome_default_intervals: bool = False,
      gvcf_external_header: Optional[str] = None,
      gvcf_sample_names: Optional[List[str]] = None,
      gvcf_info_to_keep: Optional[Collection[str]] = None,
      gvcf_reference_entry_fields_to_keep: Optional[
          Collection[str]
      ] = None,
      call_fields: Collection[str] = ["PGT"],
      branch_factor: int = VariantDatasetCombiner._default_branch_factor,
      target_records: int = VariantDatasetCombiner._default_target_records,
      gvcf_batch_size: Optional[int] = None,
      batch_size: Optional[int] = None,
      reference_genome: Union[str, hl.ReferenceGenome] = "default",
      contig_recoding: Optional[Dict[str, str]] = None,
      force: bool = False,
  ) -> VariantDatasetCombiner:
      """Create a new :class:`.VariantDatasetCombiner` or load one from `save_path`."""
      if not (gvcf_paths or vds_paths):
          raise ValueError(
              "at least one  of 'gvcf_paths' or 'vds_paths' must be nonempty"
          )
      if gvcf_paths is None:
          gvcf_paths = []
      if vds_paths is None:
          vds_paths = []
      if vds_sample_counts is not None and len(vds_paths) != len(
          vds_sample_counts
      ):
          raise ValueError(
              "'vds_paths' and 'vds_sample_counts' (if present) must have the same length "
              f"{len(vds_paths)} != {len(vds_sample_counts)}"
          )
      if (gvcf_sample_names is None) != (gvcf_external_header is None):
          raise ValueError(
              "both 'gvcf_sample_names' and 'gvcf_external_header' must be set or unset"
          )
      if gvcf_sample_names is not None and len(
          gvcf_sample_names
      ) != len(gvcf_paths):
          raise ValueError(
              "'gvcf_sample_names' and 'gvcf_paths' must have the same length "
              f"{len(gvcf_sample_names)} != {len(gvcf_paths)}"
          )
  
      if batch_size is None:
          if gvcf_batch_size is None:
              gvcf_batch_size = (
                  VariantDatasetCombiner._default_gvcf_batch_size
              )
          else:
              pass
      else:
          if gvcf_batch_size is None:
              warning(
                  "The batch_size parameter is deprecated. "
                  "The batch_size parameter will be removed in a future version of Hail. "
                  "Please use gvcf_batch_size instead."
              )
              gvcf_batch_size = batch_size
          else:
              raise ValueError(
                  "Specify only one of batch_size and gvcf_batch_size. "
                  f"Received {batch_size} and {gvcf_batch_size}."
              )
      del batch_size
  
      def maybe_load_from_saved_path(
          save_path: str,
      ) -> Optional[VariantDatasetCombiner]:
          if force:
              return None
          fs = hl.current_backend().fs
          if fs.exists(save_path):
              try:
                  combiner = load_combiner(save_path)
                  warning(
                      f"found existing combiner plan at {save_path}, using it"
                  )
                  # we overwrite these values as they are serialized, but not part of the
                  # hash for an autogenerated name and we want users to be able to overwrite
                  # these when resuming a combine (a common reason to need to resume a combine
                  # is a failure due to branch factor being too large)
                  combiner._branch_factor = branch_factor
                  combiner._target_records = target_records
                  combiner._gvcf_batch_size = gvcf_batch_size
                  return combiner
              except (ValueError, TypeError, OSError, KeyError) as e:
                  warning(
                      f"file exists at {save_path}, but it is not a valid combiner plan, overwriting\n"
                      f"    caused by: {e}"
                  )
          return None
  
      # We do the first save_path check now after validating the arguments
      if save_path is not None:
          saved_combiner = maybe_load_from_saved_path(save_path)
          if saved_combiner is not None:
              return saved_combiner
  
      if len(gvcf_paths) > 0:
          n_partition_args = (
              int(intervals is not None)
              + int(import_interval_size is not None)
              + int(use_genome_default_intervals)
              + int(use_exome_default_intervals)
          )
  
          if n_partition_args == 0:
              raise ValueError(
                  "'new_combiner': require one argument from 'intervals', 'import_interval_size', "
                  "'use_genome_default_intervals', or 'use_exome_default_intervals' to choose GVCF partitioning"
              )
  
          if n_partition_args > 1:
              warning(
                  "'new_combiner': multiple colliding arguments found from 'intervals', 'import_interval_size', "
                  "'use_genome_default_intervals', or 'use_exome_default_intervals'."
                  "\n  The argument found first in the list in this warning will be used, and others ignored."
              )
  
          if intervals is not None:
              pass
          elif import_interval_size is not None:
              intervals = calculate_even_genome_partitioning(
                  reference_genome, import_interval_size
              )
          elif use_genome_default_intervals:
              size = VariantDatasetCombiner.default_genome_interval_size
              intervals = calculate_even_genome_partitioning(
                  reference_genome, size
              )
          elif use_exome_default_intervals:
              size = VariantDatasetCombiner.default_exome_interval_size
              intervals = calculate_even_genome_partitioning(
                  reference_genome, size
              )
          assert intervals is not None
      else:
          intervals = []
  
      if isinstance(reference_genome, str):
          reference_genome = hl.get_reference(reference_genome)
  
      # we need to compute the type that the combiner will have, this will allow us to read matrix
      # tables quickly, especially in an asynchronous environment like query on batch where typing
      # a read uses a blocking round trip.
      vds = None
      gvcf_type = None
      if vds_paths:
          # sync up gvcf_reference_entry_fields_to_keep and they reference entry types from the VDS
          vds = hl.vds.read_vds(
              vds_paths[0], _warn_no_ref_block_max_length=False
          )
          vds_ref_entry = set(vds.reference_data.entry) - {"END"}
          if (
              gvcf_reference_entry_fields_to_keep is not None
              and vds_ref_entry != gvcf_reference_entry_fields_to_keep
          ):
              warning(
                  "Mismatch between 'gvcf_reference_entry_fields' to keep and VDS reference data "
                  "entry types. Overwriting with reference entry fields from supplied VDS.\n"
                  f"    VDS reference entry fields      : {sorted(vds_ref_entry)}\n"
                  f"    requested reference entry fields: {sorted(gvcf_reference_entry_fields_to_keep)}"
              )
          gvcf_reference_entry_fields_to_keep = vds_ref_entry
  
          # sync up call_fields and call fields present in the VDS
          all_entry_types = chain(
              vds.reference_data._type.entry_type.items(),
              vds.variant_data._type.entry_type.items(),
          )
          vds_call_fields = {
              name for name, typ in all_entry_types if typ == hl.tcall
          } - {"LGT", "GT"}
          if "LPGT" in vds_call_fields:
              vds_call_fields = (vds_call_fields - {"LPGT"}) | {"PGT"}
          if set(call_fields) != vds_call_fields:
              warning(
                  "Mismatch between 'call_fields' and VDS call fields. "
                  "Overwriting with call fields from supplied VDS.\n"
                  f"    VDS call fields      : {sorted(vds_call_fields)}\n"
                  f"    requested call fields: {sorted(call_fields)}\n"
              )
          call_fields = vds_call_fields
  
      if gvcf_paths:
          mt = hl.import_vcf(
              gvcf_paths[0],
              header_file=gvcf_external_header,
              force_bgz=True,
              array_elements_required=False,
              reference_genome=reference_genome,
              contig_recoding=contig_recoding,
          )
          gvcf_type = mt._type
          if gvcf_reference_entry_fields_to_keep is None:
              rmt = mt.filter_rows(hl.is_defined(mt.info.END))
              gvcf_reference_entry_fields_to_keep = (
                  defined_entry_fields(rmt, 100_000)
                  - {"GT", "PGT", "PL"}
              )
          if vds is None:
              vds = transform_gvcf(
                  mt._key_rows_by_assert_sorted("locus"),
                  gvcf_reference_entry_fields_to_keep,
                  gvcf_info_to_keep,
              )
      dataset_type = CombinerOutType(
          reference_type=vds.reference_data._type,
          variant_type=vds.variant_data._type,
      )
  
      if save_path is None:
          sha = hashlib.sha256()
          sha.update(output_path.encode())
          sha.update(temp_path.encode())
          sha.update(str(reference_genome).encode())
          sha.update(str(dataset_type).encode())
          if gvcf_type is not None:
              sha.update(str(gvcf_type).encode())
          for path in vds_paths:
              sha.update(path.encode())
          for path in gvcf_paths:
              sha.update(path.encode())
          if gvcf_external_header is not None:
              sha.update(gvcf_external_header.encode())
          if gvcf_sample_names is not None:
              for name in gvcf_sample_names:
                  sha.update(name.encode())
          if gvcf_info_to_keep is not None:
              for kept_info in sorted(gvcf_info_to_keep):
                  sha.update(kept_info.encode())
          if gvcf_reference_entry_fields_to_keep is not None:
              for field in sorted(gvcf_reference_entry_fields_to_keep):
                  sha.update(field.encode())
          for call_field in sorted(call_fields):
              sha.update(call_field.encode())
          if contig_recoding is not None:
              for key, value in sorted(contig_recoding.items()):
                  sha.update(key.encode())
                  sha.update(value.encode())
          for interval in intervals:
              sha.update(str(interval).encode())
          digest = sha.hexdigest()
          name = f"vds-combiner-plan_{digest}_{hl.__pip_version__}.json"
          save_path = os.path.join(temp_path, "combiner-plans", name)
          saved_combiner = maybe_load_from_saved_path(save_path)
          if saved_combiner is not None:
              return saved_combiner
          warning(f"generated combiner save path of {save_path}")
  
      if vds_sample_counts:
          vdses = [
              VDSMetadata(path, n_samples)
              for path, n_samples in zip(vds_paths, vds_sample_counts)
          ]
      else:
          vdses = []
          for path in vds_paths:
              vds = hl.vds.read_vds(
                  path,
                  _assert_reference_type=dataset_type.reference_type,
                  _assert_variant_type=dataset_type.variant_type,
                  _warn_no_ref_block_max_length=False,
              )
              n_samples = vds.n_samples()
              vdses.append(VDSMetadata(path, n_samples))
  
      vdses.sort(key=lambda x: x.n_samples, reverse=True)
  
      combiner = VariantDatasetCombiner(
          save_path=save_path,
          output_path=output_path,
          temp_path=temp_path,
          reference_genome=reference_genome,
          dataset_type=dataset_type,
          branch_factor=branch_factor,
          target_records=target_records,
          gvcf_batch_size=gvcf_batch_size,
          contig_recoding=contig_recoding,
          call_fields=call_fields,
          vdses=vdses,
          gvcfs=gvcf_paths,
          gvcf_import_intervals=intervals,
          gvcf_external_header=gvcf_external_header,
          gvcf_sample_names=gvcf_sample_names,
          gvcf_info_to_keep=gvcf_info_to_keep,
          gvcf_reference_entry_fields_to_keep=gvcf_reference_entry_fields_to_keep,
      )
      combiner._raise_if_output_exists()
      return combiner
  
imports: import collections
  import hashlib
  import json
  import os
  import sys
  import uuid
  from itertools import chain
  from math import floor, log
  from typing import Collection, Dict, List, NamedTuple, Optional, Union
  import hail
  from hail.expr import HailType, tmatrix
  from hail.utils import FatalError, Interval
  from hail.utils.java import info, warning
  from .combine import combine_variant_datasets, transform_gvcf, defined_entry_fields, make_variant_stream, make_reference_stream, combine_r, calculate_even_genome_partitioning, calculate_new_intervals, combine
  from ..variant_dataset import VariantDataset
label_desc: Use image node_id 1 for calling a global function with example usage: new_combiner() and returns: combiner, None, None, saved_combiner, saved_combiner, combiner