code: def truncate_reference_blocks(
      ds,
      *,
      max_ref_block_base_pairs=None,
      ref_block_winsorize_fraction=None,
  ):
      """Cap reference blocks at a maximum length in order to permit faster interval filtering.
  
      Examples
      --------
      Truncate reference blocks to 5 kilobases:
  
      >>> vds2 = hl.vds.truncate_reference_blocks(vds, max_ref_block_base_pairs=5000) # doctest: +SKIP
  
      Truncate the longest 1% of reference blocks to the length of the 99th percentile block:
  
      >>> vds2 = hl.vds.truncate_reference_blocks(vds, ref_block_winsorize_fraction=0.01) # doctest: +SKIP
  
      Notes
      -----
      After this function has been run, the reference blocks have a known maximum length `ref_block_max_length`,
      stored in the global fields, which permits :func:`.vds.filter_intervals` to filter to intervals of the reference
      data by reading `ref_block_max_length` bases ahead of each interval. This allows narrow interval queries
      to run in roughly O(data kept) work rather than O(all reference data) work.
  
      It is also possible to patch an existing VDS to store the max reference block length with :func:`.vds.store_ref_block_max_length`.
  
      See Also
      --------
      :func:`.vds.store_ref_block_max_length`.
  
      Parameters
      ----------
      vds : :class:`.VariantDataset` or :class:`.MatrixTable`
      max_ref_block_base_pairs
          Maximum size of reference blocks, in base pairs.
      ref_block_winsorize_fraction
          Fraction of reference block length distribution to truncate / winsorize.
  
      Returns
      -------
      :class:`.VariantDataset` or :class:`.MatrixTable`
      """
      if isinstance(ds, VariantDataset):
          rd = ds.reference_data
      else:
          rd = ds
  
      fd_name = hl.vds.VariantDataset.ref_block_max_length_field
      if fd_name in rd.globals:
          rd = rd.drop(fd_name)
  
      if (
          int(ref_block_winsorize_fraction is None)
          + int(max_ref_block_base_pairs is None)
          != 1
      ):
          raise ValueError(
              'truncate_reference_blocks: require exactly one of "max_ref_block_base_pairs", "ref_block_winsorize_fraction"'
          )
  
      if ref_block_winsorize_fraction is not None:
          assert (
              ref_block_winsorize_fraction > 0
              and ref_block_winsorize_fraction < 1
          ), 'truncate_reference_blocks: "ref_block_winsorize_fraction" must be between 0 and 1 (e.g. 0.01 to truncate the top 1% of reference blocks)'
          if ref_block_winsorize_fraction > 0.1:
              warning(
                  f"'truncate_reference_blocks': ref_block_winsorize_fraction of {ref_block_winsorize_fraction} will lead to significant data duplication,"
                  f" recommended values are <0.05."
              )
          max_ref_block_base_pairs = rd.aggregate_entries(
              hl.agg.approx_quantiles(
                  rd.END - rd.locus.position + 1,
                  1 - ref_block_winsorize_fraction,
                  k=200,
              )
          )
  
      assert (
          max_ref_block_base_pairs > 0
      ), 'truncate_reference_blocks: "max_ref_block_base_pairs" must be between greater than zero'
      info(
          f"splitting VDS reference blocks at {max_ref_block_base_pairs} base pairs"
      )
  
      rd_under_limit = rd.filter_entries(
          rd.END - rd.locus.position < max_ref_block_base_pairs
      ).localize_entries("fixed_blocks", "cols")
  
      rd_over_limit = rd.filter_entries(
          rd.END - rd.locus.position >= max_ref_block_base_pairs
      ).key_cols_by(col_idx=hl.scan.count())
      rd_over_limit = (
          rd_over_limit.select_rows()
          .select_cols()
          .key_rows_by()
          .key_cols_by()
      )
      es = rd_over_limit.entries()
      es = es.annotate(
          new_start=hl.range(
              es.locus.position, es.END + 1, max_ref_block_base_pairs
          )
      )
      es = es.explode("new_start")
      es = es.transmute(
          locus=hl.locus(
              es.locus.contig,
              es.new_start,
              reference_genome=es.locus.dtype.reference_genome,
          ),
          END=hl.min(
              es.new_start + max_ref_block_base_pairs - 1, es.END
          ),
      )
      es = es.key_by(es.locus).collect_by_key("new_blocks")
      es = es.transmute(
          moved_blocks_dict=hl.dict(
              es.new_blocks.map(
                  lambda x: (x.col_idx, x.drop("col_idx"))
              )
          )
      )
  
      joined = rd_under_limit.join(es, how="outer")
      joined = joined.transmute(
          merged_blocks=hl.range(hl.len(joined.cols)).map(
              lambda idx: hl.coalesce(
                  joined.moved_blocks_dict.get(idx),
                  joined.fixed_blocks[idx],
              )
          )
      )
      new_rd = joined._unlocalize_entries(
          entries_field_name="merged_blocks",
          cols_field_name="cols",
          col_key=list(rd.col_key),
      )
      new_rd = new_rd.annotate_globals(
          **{fd_name: max_ref_block_base_pairs}
      )
  
      if isinstance(ds, hl.vds.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 15 for calling a global function with example usage: truncate_reference_blocks(ds) and returns: new_rd, VariantDataset