code: def merge_reference_blocks(
      ds, equivalence_function, merge_functions=None
  ):
      """Merge adjacent reference blocks according to user equivalence criteria.
  
      Examples
      --------
      Coarsen GQ granularity into bins of 10 and merges blocks with the same GQ in order to
      compress reference data.
  
      >>> rd = vds.reference_data # doctest: +SKIP
      >>> vds.reference_data = rd.annotate_entries(GQ = rd.GQ - rd.GQ % 10) # doctest: +SKIP
      >>> vds2 = hl.vds.merge_reference_blocks(vds,
      ...                                      equivalence_function=lambda block1, block2: block1.GQ == block2.GQ),
      ...                                      merge_functions={'MIN_DP': 'min'}) # doctest: +SKIP
  
      Notes
      -----
      The `equivalence_function` argument expects a function from two reference blocks to a
      boolean value indicating whether they should be combined. Adjacency checks are builtin
      to the method (two reference blocks are 'adjacent' if the END of one block is one base
      before the beginning of the next).
  
      The `merge_functions`
  
      Parameters
      ----------
      ds : :class:`.VariantDataset` or :class:`.MatrixTable`
          Variant dataset or reference block matrix table.
      Returns
      -------
      :class:`.VariantDataset` or :class:`.MatrixTable`
      """
      if isinstance(ds, VariantDataset):
          rd = ds.reference_data
      else:
          rd = ds
      rd = rd.annotate_rows(
          contig_idx_row=rd.locus.contig_idx,
          start_pos_row=rd.locus.position,
      )
      rd = rd.annotate_entries(
          contig_idx=rd.contig_idx_row, start_pos=rd.start_pos_row
      )
      ht = rd.localize_entries("entries", "cols")
  
      def merge(block1, block2):
          new_fields = {"END": block2.END}
          if merge_functions:
              for k, f in merge_functions.items():
                  if isinstance(f, str):
                      f = f.lower()
                      if f == "min":
  
                          def f(b1, b2):
                              return hl.min(block1[k], block2[k])
  
                      elif f == "max":
  
                          def f(b1, b2):
                              return hl.max(block1[k], block2[k])
  
                      elif f == "sum":
  
                          def f(b1, b2):
                              return block1[k] + block2[k]
  
                      else:
                          raise ValueError(
                              f"merge_reference_blocks: unknown merge function {f!r},"
                              f" support 'min', 'max', and 'sum' in addition to custom lambdas"
                          )
                  new_value = f(block1, block2)
                  if new_value.dtype != block1[k].dtype:
                      raise ValueError(
                          f"merge_reference_blocks: merge_function for {k!r}: new type {new_value.dtype!r} "
                          f"differs from original type {block1[k].dtype!r}"
                      )
                  new_fields[k] = new_value
          return block1.annotate(**new_fields)
  
      def keep_last(t1, t2):
          e1 = t1[0]
          e2 = t2[0]
          are_adjacent = (e1.contig_idx == e2.contig_idx) & (
              e1.END + 1 == e2.start_pos
          )
          return hl.if_else(
              hl.is_defined(e1)
              & hl.is_defined(e2)
              & are_adjacent
              & equivalence_function(e1, e2),
              (merge(e1, e2), True),
              t2,
          )
  
      # approximate a scan that merges before result
      ht = ht.annotate(
          prev_block=hl.zip(
              hl.scan.array_agg(
                  lambda elt: hl.scan.fold(
                      (hl.missing(rd.entry.dtype), False),
                      lambda acc: keep_last(acc, (elt, False)),
                      keep_last,
                  ),
                  ht.entries,
              ),
              ht.entries,
          ).map(lambda tup: keep_last(tup[0], (tup[1], False)))
      )
      ht_join = ht
  
      ht = ht.key_by()
      ht = ht.select(
          to_shuffle=hl.enumerate(ht.prev_block).filter(
              lambda idx_and_elt: hl.is_defined(idx_and_elt[1])
              & idx_and_elt[1][1]
          )
      )
      ht = ht.explode("to_shuffle")
      rg = rd.locus.dtype.reference_genome
      ht = ht.transmute(
          col_idx=ht.to_shuffle[0], entry=ht.to_shuffle[1][0]
      )
      ht_shuf = ht.key_by(
          locus=hl.locus(
              hl.literal(rg.contigs)[ht.entry.contig_idx],
              ht.entry.start_pos,
              reference_genome=rg,
          )
      )
  
      ht_shuf = ht_shuf.collect_by_key("new_starts")
      # new_starts can contain multiple records for a collapsed ref block, one for each folded block.
      # We want to keep the one with the highest END
      ht_shuf = ht_shuf.select(
          moved_blocks_dict=hl.group_by(
              lambda elt: elt.col_idx, ht_shuf.new_starts
          ).map_values(
              lambda arr: arr[
                  hl.argmax(arr.map(lambda x: x.entry.END))
              ].entry.drop("contig_idx", "start_pos")
          )
      )
  
      ht_joined = ht_join.join(ht_shuf.select_globals(), "left")
  
      def merge_f(tup):
          (idx, original_entry) = tup
  
          return (
              hl.case()
              .when(
                  ~(hl.coalesce(ht_joined.prev_block[idx][1], False)),
                  hl.coalesce(
                      ht_joined.moved_blocks_dict.get(idx),
                      original_entry.drop("contig_idx", "start_pos"),
                  ),
              )
              .or_missing()
          )
  
      ht_joined = ht_joined.annotate(
          new_entries=hl.enumerate(ht_joined.entries).map(
              lambda tup: merge_f(tup)
          )
      )
      ht_joined = ht_joined.drop(
          "moved_blocks_dict",
          "entries",
          "prev_block",
          "contig_idx_row",
          "start_pos_row",
      )
      new_rd = ht_joined._unlocalize_entries(
          entries_field_name="new_entries",
          cols_field_name="cols",
          col_key=list(rd.col_key),
      )
  
      rbml = hl.vds.VariantDataset.ref_block_max_length_field
      if rbml in new_rd.globals:
          new_rd = new_rd.drop(rbml)
  
      if isinstance(ds, VariantDataset):
          return VariantDataset(
              reference_data=new_rd, variant_data=ds.variant_data
          )
      return new_rd
  
imports: from typing import Sequence
  import hail
  from hail import ir
  from hail.expr import expr_any, expr_array, expr_interval, expr_locus, expr_str, expr_bool
  from hail.matrixtable import MatrixTable
  from hail.methods.misc import require_first_key_field_locus
  from hail.table import Table
  from hail.typecheck import sequenceof, typecheck, nullable, oneof, enumeration, func_spec, dictof
  from hail.utils.java import Env, info, warning
  from hail.utils.misc import divide_null, new_temp_file, wrap_to_list
  from hail.vds.variant_dataset import VariantDataset
label_desc: Use image node_id 16 for calling a global function with example usage: merge_reference_blocks(ds, equivalence_function, merge_functions) and returns: new_rd, block1, hl, hl, VariantDataset, hl, hl, unknown