code: def test_ridge_reducer_transform_with_cov(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
      )
      sample_blocks = __get_sample_blocks(indexdf)
      headers = [
          r.header
          for r in blockdf.filter(
              f'header_block = "{testBlock}" AND sample_block= {testGroup}'
          )
          .orderBy("sort_key")
          .select("header")
          .collect()
      ]
  
      C_in = std_covdf.loc[ids, :].values
      X_in = X0[headers].loc[ids, :].values
      X_in_cov = np.column_stack([C_in, X_in])
      C_out = std_covdf.drop(ids, axis="rows").values
      X_out = X0[headers].drop(ids, axis="rows").values
      X_out_cov = np.column_stack((C_out, X_out))
      Y_out = std_labeldf.drop(ids, axis="rows").values
  
      XtX_out_cov = X_out_cov.T @ X_out_cov
      XtY_out_cov = X_out_cov.T @ Y_out
      diags_cov = [
          np.concatenate(
              [
                  np.ones(n_cov),
                  np.ones(XtX_out_cov.shape[1] - n_cov) * a,
              ]
          )
          for a in alphas
      ]
      B_cov = np.column_stack(
          [
              (np.linalg.inv(XtX_out_cov + np.diag(d)) @ XtY_out_cov)
              for d in diags_cov
          ]
      )
      X1_in_cov = X_in_cov @ B_cov
  
      stack = RidgeReduction(
          blockdf,
          labeldf,
          sample_blocks,
          covdf,
          add_intercept=False,
          alphas=alphas,
      )
      level1df_cov = stack.fit_transform()
  
      columns = ["alpha", "label", "values"]
      rows_cov = (
          level1df_cov.filter(
              f'header LIKE "%{testBlock}%" AND sample_block= {testGroup}'
          )
          .select(*columns)
          .collect()
      )
      outdf_cov = pd.DataFrame(rows_cov, columns=columns)
      X1_in_stack_cov = np.column_stack(outdf_cov["values"])
  
      assert np.allclose(X1_in_stack_cov, X1_in_cov)
  
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 9 for calling a global function with example usage: test_ridge_reducer_transform_with_cov(spark) without return types