code: def index_column(
      retrieval_config: Dict[str, Any],
      col_name: str,
      dataset_cols: Dict[str, Series],
      backend: "Backend",
      split_col: Optional[Series] = None,
  ) -> Tuple[RetrievalModel, str]:
      """Indexes a column for sample retrieval via embedding index lookup.
  
      This function indexes a column and saves the index artifact to disk. If an index name is provided as part of the
      `retrieval_config`, then the index in the ludwig cache with the corresponding name will be loaded instead of being
      built from scratch.
  
      To prevent data leakage, a split column must be provided. This ensures that the retrieval model only ever fetches
      samples from the training set.
  
      To ensure that the index is usable even if the original DataFrame is not available, the columns used to build the
      index are stored as part of the index.
  
      All operations in this function are performed on pandas objects, which means that you may run out of memory if your
      dataset is large.
  
      Args:
          retrieval_config (Dict[str, Any]): The retrieval config from the config object.
          col_name (str): The name of the column to index.
          dataset_cols (Dict[str, Series]): A dictionary mapping column names to their corresponding Series. `col_name`
              must be a key in this dictionary. These columns are stored as part of the index to ensure that the index
              is usable even if the original DataFrame is not available.
          df_engine (DataFrameEngine): The engine used to compute the columns into pandas objects.
          split_col (Optional[Series]): A column that indicates whether a sample is part of the training set. A sample
              is in the training set if the value in this column is 0.
      Returns:
          Tuple[RetrievalModel, str]: A tuple containing the retrieval model and the name of the index.
      """
      retrieval_model = get_retrieval_model(
          retrieval_config["type"],
          model_name=retrieval_config["model_name"],
      )
  
      index_name = retrieval_config["index_name"]
      index_cache_directory = os.path.join(
          get_default_cache_location(), "index"
      )
      if not path_exists(index_cache_directory):
          makedirs(index_cache_directory, exist_ok=True)
  
      if index_name is None:
          if split_col is None:
              raise ValueError(
                  "split column must be provided if using retrieval"
              )
          split_col = backend.df_engine.compute(split_col).astype(int)
  
          # TODO(geoffrey): add support for Dask DataFrames
          df = pd.DataFrame(
              {
                  name: backend.df_engine.compute(col)
                  for name, col in dataset_cols.items()
              }
          )
          df = df[
              split_col == 0
          ]  # Ensures that the index is only built on the training set
  
          # Even if index name is not provided, we still want to check if an index for this df already exists in cache
          # If it does, load it and return immediately
          index_hash = df_checksum(df)
          index_name = f"embedding_index_{index_hash}"
          if path_exists(
              os.path.join(index_cache_directory, index_name)
          ):
              logger.info(
                  f"Index for this DataFrame with name '{index_name}' already exists. "
                  f"Loading index from '{index_cache_directory}'"
              )
              retrieval_model.load_index(
                  index_name, cache_directory=index_cache_directory
              )
              return retrieval_model, index_name
  
          # Build index if index name is not provided and index for this df does not already exist in cache
          retrieval_model.create_dataset_index(
              df, backend, columns_to_index=[col_name]
          )
          logger.info(
              f"Saving index to cache directory '{index_cache_directory}' with name '{index_name}'"
          )
          retrieval_model.save_index(
              index_name, cache_directory=index_cache_directory
          )
      else:
          logger.info(
              f"Loading index from cache directory '{index_cache_directory}' with name '{index_name}'"
          )
          retrieval_model.load_index(
              index_name, cache_directory=index_cache_directory
          )
      return retrieval_model, index_name
  
file_path: ludwig/ludwig/data/prompt.py
imports: import json
  import logging
  import os
  import string
  from typing import Any, Dict, List, Optional, Set, Tuple, Type, TYPE_CHECKING
  import pandas
  from ludwig.models.retrieval import df_checksum, get_retrieval_model, RetrievalModel
  from ludwig.utils.fs_utils import get_default_cache_location, makedirs, path_exists
  from ludwig.utils.types import DataFrame, Series
label_desc: Use image node_id 1 for calling a global function with example usage: index_column(retrieval_config, col_name, dataset_cols, backend, split_col) and returns: retrieval_model,  index_name, retrieval_model,  index_name