code: def _get_instance_param_map_recursively(
      instance, level, uid_to_indexed_name_map
  ):
      from pyspark.ml.param import Params
      from pyspark.ml.pipeline import Pipeline
  
      param_map = _get_param_map(instance)
      expanded_param_map = {}
  
      is_pipeline = isinstance(instance, Pipeline)
      is_parameter_search_estimator = _is_parameter_search_estimator(
          instance
      )
  
      logged_param_name_prefix = (
          ""
          if level == 0
          else uid_to_indexed_name_map[instance.uid] + "."
      )
  
      for param_name, param_value in param_map.items():
          logged_param_name = logged_param_name_prefix + param_name
  
          if is_pipeline and param_name == "stages":
              expanded_param_map[logged_param_name] = [
                  uid_to_indexed_name_map[stage.uid]
                  for stage in instance.getStages()
              ]
              for stage in instance.getStages():
                  stage_param_map = _get_instance_param_map_recursively(
                      stage, level + 1, uid_to_indexed_name_map
                  )
                  expanded_param_map.update(stage_param_map)
          elif (
              is_parameter_search_estimator
              and param_name == "estimator"
          ):
              expanded_param_map[
                  logged_param_name
              ] = uid_to_indexed_name_map[param_value.uid]
              # skip log estimator's nested params because they will be logged as JSON artifact,
              # and they will be logged in nested runs as well.
          elif (
              is_parameter_search_estimator
              and param_name == "estimatorParamMaps"
          ):
              # this param will be saved as JSON format artifact.
              pass
          elif isinstance(param_value, Params):
              # handle the case param value type inherits `pyspark.ml.param.Params`
              # e.g. param like
              # `OneVsRest.classifier`/`CrossValidator.evaluator`
              expanded_param_map[
                  logged_param_name
              ] = uid_to_indexed_name_map[param_value.uid]
              internal_param_map = _get_instance_param_map_recursively(
                  param_value, level + 1, uid_to_indexed_name_map
              )
              expanded_param_map.update(internal_param_map)
          else:
              expanded_param_map[logged_param_name] = param_value
  
      return expanded_param_map
  
imports: import json
  import logging
  import os
  import sys
  import traceback
  import weakref
  from collections import OrderedDict, defaultdict, namedtuple
  from itertools import zip_longest
  from urllib.parse import urlparse
  import numpy
  import mlflow
  from mlflow.data.code_dataset_source import CodeDatasetSource
  from mlflow.data.spark_dataset import SparkDataset
  from mlflow.entities import Metric, Param
  from mlflow.entities.dataset_input import DatasetInput
  from mlflow.entities.input_tag import InputTag
  from mlflow.exceptions import MlflowException
  from mlflow.tracking.client import MlflowClient
  from mlflow.utils import _chunk_dict, _get_fully_qualified_class_name, _inspect_original_var_name, _truncate_dict
  from mlflow.utils.autologging_utils import INPUT_EXAMPLE_SAMPLE_ROWS, _get_new_training_session_class, autologging_integration, get_method_call_arg_value, resolve_input_example_and_signature, safe_patch
  from mlflow.utils.file_utils import TempDir
  from mlflow.utils.mlflow_tags import MLFLOW_AUTOLOGGING, MLFLOW_DATASET_CONTEXT, MLFLOW_PARENT_RUN_ID
  from mlflow.utils.rest_utils import MlflowHostCreds, augmented_raise_for_status, http_request
  from mlflow.utils.time import get_current_time_millis
  from mlflow.utils.validation import MAX_ENTITY_KEY_LENGTH, MAX_PARAM_VAL_LENGTH, MAX_PARAMS_TAGS_PER_BATCH
label_desc: Use image node_id 15 for calling a global function with example usage: _get_instance_param_map_recursively(instance, level, uid_to_indexed_name_map) and returns: expanded_param_map