code: def read_csv(
      path: Union[str, List[str]],
      sep: str = ",",
      header: Union[str, int, None] = "infer",
      names: Optional[Union[str, List[str]]] = None,
      index_col: Optional[Union[str, List[str]]] = None,
      usecols: Optional[
          Union[List[int], List[str], Callable[[str], bool]]
      ] = None,
      dtype: Optional[
          Union[str, Dtype, Dict[str, Union[str, Dtype]]]
      ] = None,
      nrows: Optional[int] = None,
      parse_dates: bool = False,
      quotechar: Optional[str] = None,
      escapechar: Optional[str] = None,
      comment: Optional[str] = None,
      encoding: Optional[str] = None,
      **options: Any,
  ) -> Union[DataFrame, Series]:
      """Read CSV (comma-separated) file into DataFrame or Series.
  
      Parameters
      ----------
      path : str or list
          Path(s) of the CSV file(s) to be read.
      sep : str, default ‘,’
          Delimiter to use. Non empty string.
      header : int, default ‘infer’
          Whether to use the column names, and the start of the data.
          Default behavior is to infer the column names: if no names are passed
          the behavior is identical to `header=0` and column names are inferred from
          the first line of the file, if column names are passed explicitly then
          the behavior is identical to `header=None`. Explicitly pass `header=0` to be
          able to replace existing names
      names : str or array-like, optional
          List of column names to use. If file contains no header row, then you should
          explicitly pass `header=None`. Duplicates in this list will cause an error to be issued.
          If a string is given, it should be a DDL-formatted string in Spark SQL, which is
          preferred to avoid schema inference for better performance.
      index_col: str or list of str, optional, default: None
          Index column of table in Spark.
      usecols : list-like or callable, optional
          Return a subset of the columns. If list-like, all elements must either be
          positional (i.e. integer indices into the document columns) or strings that
          correspond to column names provided either by the user in names or inferred
          from the document header row(s).
          If callable, the callable function will be evaluated against the column names,
          returning names where the callable function evaluates to `True`.
      dtype : Type name or dict of column -> type, default None
          Data type for data or columns. E.g. {‘a’: np.float64, ‘b’: np.int32} Use str or object
          together with suitable na_values settings to preserve and not interpret dtype.
      nrows : int, default None
          Number of rows to read from the CSV file.
      parse_dates : boolean or list of ints or names or list of lists or dict, default `False`.
          Currently only `False` is allowed.
      quotechar : str (length 1), optional
          The character used to denote the start and end of a quoted item. Quoted items can include
          the delimiter and it will be ignored.
      escapechar : str (length 1), default None
          One-character string used to escape other characters.
      comment: str, optional
          Indicates the line should not be parsed.
      encoding: str, optional
          Indicates the encoding to read file
      options : dict
          All other options passed directly into Spark's data source.
  
      Returns
      -------
      DataFrame or Series
  
      See Also
      --------
      DataFrame.to_csv : Write DataFrame to a comma-separated values (csv) file.
  
      Examples
      --------
      >>> ps.read_csv('data.csv')  # doctest: +SKIP
  
      Load multiple CSV files as a single DataFrame:
  
      >>> ps.read_csv(['data-01.csv', 'data-02.csv'])  # doctest: +SKIP
      """
      # For latin-1 encoding is same as iso-8859-1, that's why its mapped to iso-8859-1.
      encoding_mapping = {"latin-1": "iso-8859-1"}
  
      if (
          "options" in options
          and isinstance(options.get("options"), dict)
          and len(options) == 1
      ):
          options = options.get("options")
  
      if parse_dates is not False:
          raise ValueError(
              "parse_dates can only be `False`: %s" % parse_dates
          )
  
      if usecols is not None and not callable(usecols):
          usecols = list(usecols)  # type: ignore[assignment]
  
      if usecols is None or callable(usecols) or len(usecols) > 0:
          reader = default_session().read
          reader.option("inferSchema", True)
          reader.option("sep", sep)
  
          if header == "infer":
              header = 0 if names is None else None
          if header == 0:
              reader.option("header", True)
          elif header is None:
              reader.option("header", False)
          else:
              raise ValueError(
                  "Unknown header argument {}".format(header)
              )
  
          if quotechar is not None:
              reader.option("quote", quotechar)
          if escapechar is not None:
              reader.option("escape", escapechar)
  
          if comment is not None:
              if not isinstance(comment, str) or len(comment) != 1:
                  raise ValueError(
                      "Only length-1 comment characters supported"
                  )
              reader.option("comment", comment)
  
          reader.options(**options)
  
          if encoding is not None:
              reader.option(
                  "encoding", encoding_mapping.get(encoding, encoding)
              )
  
          column_labels: Dict[Any, str]
          if isinstance(names, str):
              sdf = reader.schema(names).csv(path)
              column_labels = {col: col for col in sdf.columns}
          else:
              sdf = reader.csv(path)
              if is_list_like(names):
                  names = list(names)
                  if len(set(names)) != len(names):
                      raise ValueError("Found non-unique column index")
                  if len(names) != len(sdf.columns):
                      raise ValueError(
                          "The number of names [%s] does not match the number "
                          "of columns [%d]. Try names by a Spark SQL DDL-formatted "
                          "string." % (len(sdf.schema), len(names))
                      )
                  column_labels = dict(zip(names, sdf.columns))
              elif header is None:
                  column_labels = dict(enumerate(sdf.columns))
              else:
                  column_labels = {col: col for col in sdf.columns}
  
          if usecols is not None:
              missing: List[Union[int, str]]
              if callable(usecols):
                  column_labels = {
                      label: col
                      for label, col in column_labels.items()
                      if usecols(label)
                  }
                  missing = []
              elif all(isinstance(col, int) for col in usecols):
                  usecols_ints = cast(List[int], usecols)
                  new_column_labels = {
                      label: col
                      for i, (label, col) in enumerate(
                          column_labels.items()
                      )
                      if i in usecols_ints
                  }
                  missing = [
                      col
                      for col in usecols_ints
                      if (
                          col >= len(column_labels)
                          or list(column_labels)[col]
                          not in new_column_labels
                      )
                  ]
                  column_labels = new_column_labels
              elif all(isinstance(col, str) for col in usecols):
                  new_column_labels = {
                      label: col
                      for label, col in column_labels.items()
                      if label in usecols
                  }
                  missing = [
                      col
                      for col in usecols
                      if col not in new_column_labels
                  ]
                  column_labels = new_column_labels
              else:
                  raise ValueError(
                      "'usecols' must either be list-like of all strings, "
                      "all unicode, all integers or a callable."
                  )
              if len(missing) > 0:
                  raise ValueError(
                      "Usecols do not match columns, columns expected but not "
                      "found: %s" % missing
                  )
  
              if len(column_labels) > 0:
                  sdf = sdf.select(
                      [
                          scol_for(sdf, col)
                          for col in column_labels.values()
                      ]
                  )
              else:
                  sdf = default_session().createDataFrame(
                      [], schema=StructType()
                  )
      else:
          sdf = default_session().createDataFrame(
              [], schema=StructType()
          )
          column_labels = {}
  
      if nrows is not None:
          sdf = sdf.limit(nrows)
  
      index_spark_column_names: List[str]
      index_names: List[Label]
      if index_col is not None:
          if isinstance(index_col, (str, int)):
              index_col = [index_col]
          for col in index_col:
              if col not in column_labels:
                  raise KeyError(col)
          index_spark_column_names = [
              column_labels[col] for col in index_col
          ]
          index_names = [(col,) for col in index_col]
          column_labels = {
              label: col
              for label, col in column_labels.items()
              if label not in index_col
          }
      else:
          log_advice(
              "If `index_col` is not specified for `read_csv`, "
              "the default index is attached which can cause additional overhead."
          )
          index_spark_column_names = []
          index_names = []
  
      psdf: DataFrame = DataFrame(
          InternalFrame(
              spark_frame=sdf,
              index_spark_columns=[
                  scol_for(sdf, col) for col in index_spark_column_names
              ],
              index_names=index_names,
              column_labels=[
                  label if is_name_like_tuple(label) else (label,)
                  for label in column_labels
              ],
              data_spark_columns=[
                  scol_for(sdf, col) for col in column_labels.values()
              ],
          )
      )
  
      if dtype is not None:
          if isinstance(dtype, dict):
              for col, tpe in dtype.items():
                  psdf[col] = psdf[col].astype(tpe)
          else:
              for col in psdf.columns:
                  psdf[col] = psdf[col].astype(dtype)
  
      return psdf
  
imports: from typing import Any, Callable, Dict, List, Optional, Set, Sized, Tuple, Type, Union, cast, no_type_check
  from collections.abc import Iterable
  from datetime import tzinfo
  from functools import reduce
  from io import BytesIO
  import json
  import warnings
  import numpy
  import pandas
  from pandas.api.types import is_datetime64_dtype, is_list_like
  from pandas.tseries.offsets import DateOffset
  import pyarrow
  import pyarrow.parquet
  from pyspark.sql import functions, Column
  from pyspark.sql.functions import pandas_udf
  from pyspark.sql.types import ByteType, ShortType, IntegerType, LongType, FloatType, DoubleType, BooleanType, TimestampType, TimestampNTZType, DecimalType, StringType, DateType, StructType, DataType
  from pyspark.sql.dataframe import DataFrame
  from pyspark import pandas
  from pyspark.pandas._typing import Axis, Dtype, Label, Name
  from pyspark.pandas.base import IndexOpsMixin
  from pyspark.pandas.utils import align_diff_frames, default_session, is_name_like_tuple, is_name_like_value, name_like_string, same_anchor, scol_for, validate_axis, log_advice
  from pyspark.pandas.frame import DataFrame, _reduce_spark_multi
  from pyspark.pandas.internal import InternalFrame, DEFAULT_SERIES_NAME, HIDDEN_COLUMNS, SPARK_INDEX_NAME_FORMAT
  from pyspark.pandas.series import Series, first_series
  from pyspark.pandas.spark.utils import as_nullable_spark_type, force_decimal_precision_scale
  from pyspark.pandas.indexes import Index, DatetimeIndex, TimedeltaIndex
  from pyspark.pandas.indexes.multi import MultiIndex
  from pyspark.sql.utils import get_column_class
label_desc: Use image node_id 3 for calling a global function with example usage: read_csv(path, sep, header, names, index_col, usecols, dtype, nrows, parse_dates, quotechar, escapechar, comment, encoding) and returns: psdf