code: def test_reduce_normal_eqn(spark):
      indexdf = spark.read.parquet(
          f"{data_root}/groupedIDs.snappy.parquet"
      )
      blockdf = spark.read.parquet(
          f"{data_root}/blockedGT.snappy.parquet"
      )
      testGroup = "0"
      testBlock = "chr_1_block_0"
      ids = (
          indexdf.filter(f"sample_block = {testGroup}")
          .select("sample_ids")
          .head()
          .sample_ids
      )
      headers = [
          r.header
          for r in blockdf.filter(
              f'header_block = "{testBlock}" AND sample_block = {testGroup}'
          )
          .orderBy("sort_key")
          .select("header")
          .collect()
      ]
  
      X_out = X0[headers].drop(ids, axis="rows")
      Y_out = labeldf.drop(ids, axis="rows")
  
      XtX_out = X_out.to_numpy().T @ X_out.to_numpy()
      XtY_out = X_out.to_numpy().T @ Y_out.to_numpy()
  
      sample_blocks = __get_sample_blocks(indexdf)
      map_key_pattern = ["header_block", "sample_block"]
      reduce_key_pattern = ["header_block", "header"]
      map_udf = pandas_udf(
          lambda key, pdf: map_normal_eqn(
              key,
              map_key_pattern,
              pdf,
              labeldf,
              sample_blocks,
              covdf_empty,
          ),
          normal_eqn_struct,
          PandasUDFType.GROUPED_MAP,
      )
      reduce_udf = pandas_udf(
          lambda key, pdf: reduce_normal_eqn(
              key, reduce_key_pattern, pdf
          ),
          normal_eqn_struct,
          PandasUDFType.GROUPED_MAP,
      )
  
      mapdf = blockdf.groupBy(map_key_pattern).apply(map_udf)
  
      outdf = (
          mapdf.groupBy(reduce_key_pattern)
          .apply(reduce_udf)
          .filter(
              f'header_block = "{testBlock}" AND sample_block = {testGroup}'
          )
          .orderBy("sort_key")
          .select("xtx", "xty")
          .collect()
      )
  
      XtX_out_lvl = np.array([r.xtx for r in outdf])
      XtY_out_lvl = np.array([r.xty for r in outdf])
  
      assert np.allclose(XtX_out_lvl, XtX_out) and np.allclose(
          XtY_out_lvl, XtY_out
      )
  
imports: from glow.wgr.ridge_reduction import *
  from glow.wgr.ridge_regression import *
  from glow.wgr.ridge_udfs import *
  from glow.wgr.model_functions import _fill_na_and_standardize
  import numpy
  import pandas
  from pyspark.sql.functions import *
  import math
label_desc: Use image node_id 4 for calling a global function with example usage: test_reduce_normal_eqn(spark) without return types