code: def ld_prune(
      call_expr,
      r2=0.2,
      bp_window_size=1000000,
      memory_per_core=256,
      keep_higher_maf=True,
      block_size=None,
  ):
      """Returns a maximal subset of variants that are nearly uncorrelated within each window.
  
      .. include:: ../_templates/req_diploid_gt.rst
  
      .. include:: ../_templates/req_biallelic.rst
  
      .. include:: ../_templates/req_tvariant.rst
  
      Examples
      --------
      Prune variants in linkage disequilibrium by filtering a dataset to those variants returned
      by :func:`.ld_prune`. If the dataset contains multiallelic variants, the multiallelic variants
      must be filtered out or split before being passed to :func:`.ld_prune`.
  
      >>> biallelic_dataset = dataset.filter_rows(hl.len(dataset.alleles) == 2)
      >>> pruned_variant_table = hl.ld_prune(biallelic_dataset.GT, r2=0.2, bp_window_size=500000)
      >>> filtered_ds = dataset.filter_rows(hl.is_defined(pruned_variant_table[dataset.row_key]))
  
      Notes
      -----
      This method finds a maximal subset of variants such that the squared Pearson
      correlation coefficient :math:`r^2` of any pair at most `bp_window_size`
      base pairs apart is strictly less than `r2`. Each variant is represented as
      a vector over samples with elements given by the (mean-imputed) number of
      alternate alleles. In particular, even if present, **phase information is
      ignored**. Variants that do not vary across samples are dropped.
  
      The method prunes variants in linkage disequilibrium in three stages.
  
      - The first, "local pruning" stage prunes correlated variants within each
        partition, using a local variant queue whose size is determined by
        `memory_per_core`. A larger queue may facilitate more local pruning in
        this stage. Minor allele frequency is not taken into account. The
        parallelism is the number of matrix table partitions.
  
      - The second, "global correlation" stage uses block-sparse matrix
        multiplication to compute correlation between each pair of remaining
        variants within `bp_window_size` base pairs, and then forms a graph of
        correlated variants. The parallelism of writing the locally-pruned matrix
        table as a block matrix is ``n_locally_pruned_variants / block_size``.
  
      - The third, "global pruning" stage applies :func:`.maximal_independent_set`
        to prune variants from this graph until no edges remain. This algorithm
        iteratively removes the variant with the highest vertex degree. If
        `keep_higher_maf` is true, then in the case of a tie for highest degree,
        the variant with lowest minor allele frequency is removed.
  
      Warning
      -------
      The locally-pruned matrix table and block matrix are stored as temporary files
      on persistent disk. See the warnings on `BlockMatrix.from_entry_expr` with
      regard to memory and Hadoop replication errors.
  
      Parameters
      ----------
      call_expr : :class:`.CallExpression`
          Entry-indexed call expression on a matrix table with row-indexed
          variants and column-indexed samples.
      r2 : :obj:`float`
          Squared correlation threshold (exclusive upper bound).
          Must be in the range [0.0, 1.0].
      bp_window_size: :obj:`int`
          Window size in base pairs (inclusive upper bound).
      memory_per_core : :obj:`int`
          Memory in MB per core for local pruning queue.
      keep_higher_maf: :obj:`int`
          If ``True``, break ties at each step of the global pruning stage by
          preferring to keep variants with higher minor allele frequency.
      block_size: :obj:`int`, optional
          Block size for block matrices in the second stage.
          Default given by :meth:`.BlockMatrix.default_block_size`.
  
      Returns
      -------
      :class:`.Table`
          Table of a maximal independent set of variants.
      """
      if block_size is None:
          block_size = BlockMatrix.default_block_size()
  
      if not 0.0 <= r2 <= 1:
          raise ValueError(
              f"r2 must be in the range [0.0, 1.0], found {r2}"
          )
  
      if bp_window_size < 0:
          raise ValueError(
              f"bp_window_size must be non-negative, found {bp_window_size}"
          )
  
      raise_unless_entry_indexed("ld_prune/call_expr", call_expr)
      mt = matrix_table_source("ld_prune/call_expr", call_expr)
  
      require_row_key_variant(mt, "ld_prune")
  
      #  FIXME: remove once select_entries on a field is free
      if call_expr in mt._fields_inverse:
          field = mt._fields_inverse[call_expr]
      else:
          field = Env.get_uid()
          mt = mt.select_entries(**{field: call_expr})
      mt = mt.select_rows().select_cols()
      mt = mt.distinct_by_row()
      locally_pruned_table_path = new_temp_file()
      (
          _local_ld_prune(
              require_biallelic(mt, "ld_prune"),
              field,
              r2,
              bp_window_size,
              memory_per_core,
          ).write(locally_pruned_table_path, overwrite=True)
      )
      locally_pruned_table = hl.read_table(
          locally_pruned_table_path
      ).add_index()
  
      mt = mt.annotate_rows(info=locally_pruned_table[mt.row_key])
      mt = mt.filter_rows(hl.is_defined(mt.info)).unfilter_entries()
  
      std_gt_bm = BlockMatrix.from_entry_expr(
          hl.or_else(
              (mt[field].n_alt_alleles() - mt.info.mean)
              * mt.info.centered_length_rec,
              0.0,
          ),
          block_size=block_size,
      )
      r2_bm = (std_gt_bm @ std_gt_bm.T) ** 2
  
      _, stops = hl.linalg.utils.locus_windows(
          locally_pruned_table.locus, bp_window_size
      )
  
      entries = r2_bm.sparsify_row_intervals(
          range(stops.size), stops, blocks_only=True
      ).entries(keyed=False)
      entries = entries.filter(
          (entries.entry >= r2) & (entries.i < entries.j)
      )
      entries = entries.select(
          i=hl.int32(entries.i), j=hl.int32(entries.j)
      )
  
      if keep_higher_maf:
          fields = ["mean", "locus"]
      else:
          fields = ["locus"]
  
      info = locally_pruned_table.aggregate(
          hl.agg.collect(
              locally_pruned_table.row.select("idx", *fields)
          ),
          _localize=False,
      )
      info = hl.sorted(info, key=lambda x: x.idx)
  
      entries = entries.annotate_globals(info=info)
  
      entries = entries.filter(
          (
              entries.info[entries.i].locus.contig
              == entries.info[entries.j].locus.contig
          )
          & (
              entries.info[entries.j].locus.position
              - entries.info[entries.i].locus.position
              <= bp_window_size
          )
      )
  
      if keep_higher_maf:
          entries = entries.annotate(
              i=hl.struct(
                  idx=entries.i,
                  twice_maf=hl.min(
                      entries.info[entries.i].mean,
                      2.0 - entries.info[entries.i].mean,
                  ),
              ),
              j=hl.struct(
                  idx=entries.j,
                  twice_maf=hl.min(
                      entries.info[entries.j].mean,
                      2.0 - entries.info[entries.j].mean,
                  ),
              ),
          )
  
          def tie_breaker(left, right):
              return hl.sign(right.twice_maf - left.twice_maf)
  
      else:
          tie_breaker = None
  
      variants_to_remove = hl.maximal_independent_set(
          entries.i,
          entries.j,
          keep=False,
          tie_breaker=tie_breaker,
          keyed=False,
      )
  
      locally_pruned_table = locally_pruned_table.annotate_globals(
          variants_to_remove=variants_to_remove.aggregate(
              hl.agg.collect_as_set(variants_to_remove.node.idx),
              _localize=False,
          )
      )
      return (
          locally_pruned_table.filter(
              locally_pruned_table.variants_to_remove.contains(
                  hl.int32(locally_pruned_table.idx)
              ),
              keep=False,
          )
          .select()
          .persist()
      )
  
imports: import builtins
  import itertools
  import math
  from typing import Dict, Callable, Optional, Union, Tuple, List
  import hail
  import hail
  import hail.expr.aggregators
  from hail import ir
  from hail.expr import Expression, ExpressionException, expr_float64, expr_call, expr_any, expr_numeric, expr_locus, analyze, raise_unless_entry_indexed, raise_unless_row_indexed, matrix_table_source, table_source, raise_unless_column_indexed
  from hail.expr.types import tbool, tarray, tfloat64, tint32
  from hail.genetics.reference_genome import reference_genome_type
  from hail.linalg import BlockMatrix
  from hail.matrixtable import MatrixTable
  from hail.methods.misc import require_biallelic, require_row_key_variant
  from hail.stats import LinearMixedModel
  from hail.table import Table
  from hail.typecheck import typecheck, nullable, numeric, oneof, sized_tupleof, sequenceof, enumeration, anytype
  from hail.utils import wrap_to_list, new_temp_file, FatalError
  from hail.utils.java import Env, info, warning
  from .None import pca
  from .None import relatedness
  from ..backend.spark_backend import SparkBackend
label_desc: Use image node_id 36 for calling a global function with example usage: ld_prune(call_expr, r2, bp_window_size, memory_per_core, keep_higher_maf, block_size) and returns: locally_pruned_table, hl