def _reindex_index(self, index: Optional[Union[('Index', Sequence[Any])]], fill_value: Optional[Any]) -> 'DataFrame':
    nlevels = self._internal.index_level
    assert ((nlevels <= 1) or (isinstance(index, ps.MultiIndex) and (nlevels == index.nlevels))), 'MultiIndex DataFrame can only be reindexed with a similar pandas-on-Spark MultiIndex.'
    index_columns = self._internal.index_spark_column_names
    frame = self._internal.resolved_copy.spark_frame.drop(NATURAL_ORDER_COLUMN_NAME)
    if isinstance(index, ps.Index):
        if (nlevels != index.nlevels):
            return DataFrame(index._internal.with_new_columns([])).reindex(columns=self.columns, fill_value=fill_value)
        index_names = index._internal.index_names
        scols = index._internal.index_spark_columns
        labels = index._internal.spark_frame.select([scol.alias(index_column) for (scol, index_column) in zip(scols, index_columns)])
    else:
        index = ps.Index(list(index))
        labels = index._internal.spark_frame.select(index.spark.column.alias(index_columns[0]))
        index_names = self._internal.index_names
    if (fill_value is not None):
        frame_index_columns = [verify_temp_column_name(frame, '__frame_index_column_{}__'.format(i)) for i in range(nlevels)]
        index_scols = [scol_for(frame, index_col).alias(frame_index_col) for (index_col, frame_index_col) in zip(index_columns, frame_index_columns)]
        scols = self._internal.resolved_copy.data_spark_columns
        frame = frame.select((index_scols + scols))
        temp_fill_value = verify_temp_column_name(frame, '__fill_value__')
        labels = labels.withColumn(temp_fill_value, SF.lit(fill_value))
        frame_index_scols = [scol_for(frame, col) for col in frame_index_columns]
        labels_index_scols = [scol_for(labels, col) for col in index_columns]
        joined_df = frame.join(labels, on=[(fcol == lcol) for (fcol, lcol) in zip(frame_index_scols, labels_index_scols)], how='right')
        joined_df = joined_df.select(*labels_index_scols, *[F.when(reduce((lambda c1, c2: (c1 & c2)), [(fcol.isNull() & lcol.isNotNull()) for (fcol, lcol) in zip(frame_index_scols, labels_index_scols)]), scol_for(joined_df, temp_fill_value)).otherwise(scol_for(joined_df, col)).alias(col) for col in self._internal.data_spark_column_names])
        data_fields = None
    else:
        joined_df = frame.join(labels, on=index_columns, how='right')
        data_fields = [field.copy(nullable=True) for field in self._internal.data_fields]
    sdf = joined_df.drop(NATURAL_ORDER_COLUMN_NAME)
    internal = self._internal.copy(spark_frame=sdf, index_spark_columns=[scol_for(sdf, col) for col in self._internal.index_spark_column_names], index_names=index_names, index_fields=[field.copy(name=name) for (field, name) in zip(index._internal.index_fields, self._internal.index_spark_column_names)], data_spark_columns=[scol_for(sdf, col) for col in self._internal.data_spark_column_names], data_fields=data_fields)
    return DataFrame(internal)
def _reindex_index(self, index: Optional[Union[('Index', Sequence[Any])]], fill_value: Optional[Any]) -> 'DataFrame':
    nlevels = self._internal.index_level
    assert ((nlevels <= 1) or (isinstance(index, ps.MultiIndex) and (nlevels == index.nlevels))), 'MultiIndex DataFrame can only be reindexed with a similar pandas-on-Spark MultiIndex.'
    index_columns = self._internal.index_spark_column_names
    frame = self._internal.resolved_copy.spark_frame.drop(NATURAL_ORDER_COLUMN_NAME)
    if isinstance(index, ps.Index):
        if (nlevels != index.nlevels):
            return DataFrame(index._internal.with_new_columns([])).reindex(columns=self.columns, fill_value=fill_value)
        index_names = index._internal.index_names
        scols = index._internal.index_spark_columns
        labels = index._internal.spark_frame.select([scol.alias(index_column) for (scol, index_column) in zip(scols, index_columns)])
    else:
        index = ps.Index(list(index))
        labels = index._internal.spark_frame.select(index.spark.column.alias(index_columns[0]))
        index_names = self._internal.index_names
    if (fill_value is not None):
        frame_index_columns = [verify_temp_column_name(frame, '__frame_index_column_{}__'.format(i)) for i in range(nlevels)]
        index_scols = [scol_for(frame, index_col).alias(frame_index_col) for (index_col, frame_index_col) in zip(index_columns, frame_index_columns)]
        scols = self._internal.resolved_copy.data_spark_columns
        frame = frame.select((index_scols + scols))
        temp_fill_value = verify_temp_column_name(frame, '__fill_value__')
        labels = labels.withColumn(temp_fill_value, SF.lit(fill_value))
        frame_index_scols = [scol_for(frame, col) for col in frame_index_columns]
        labels_index_scols = [scol_for(labels, col) for col in index_columns]
        joined_df = frame.join(labels, on=[(fcol == lcol) for (fcol, lcol) in zip(frame_index_scols, labels_index_scols)], how='right')
        joined_df = joined_df.select(*labels_index_scols, *[F.when(reduce((lambda c1, c2: (c1 & c2)), [(fcol.isNull() & lcol.isNotNull()) for (fcol, lcol) in zip(frame_index_scols, labels_index_scols)]), scol_for(joined_df, temp_fill_value)).otherwise(scol_for(joined_df, col)).alias(col) for col in self._internal.data_spark_column_names])
        data_fields = None
    else:
        joined_df = frame.join(labels, on=index_columns, how='right')
        data_fields = [field.copy(nullable=True) for field in self._internal.data_fields]
    sdf = joined_df.drop(NATURAL_ORDER_COLUMN_NAME)
    internal = self._internal.copy(spark_frame=sdf, index_spark_columns=[scol_for(sdf, col) for col in self._internal.index_spark_column_names], index_names=index_names, index_fields=[field.copy(name=name) for (field, name) in zip(index._internal.index_fields, self._internal.index_spark_column_names)], data_spark_columns=[scol_for(sdf, col) for col in self._internal.data_spark_column_names], data_fields=data_fields)
    return DataFrame(internal)