def METHOD_NAME(cls, op: "DataFrameDataSource"):
    df = op.outputs[0]
    raw_df = op.data
    # estimate column memory usage instead of calling df.memory_usage(deep=True)
    memory_usage = pd.Series(
        {c: estimate_pandas_size(s) for c, s in raw_df.items()}
    )
    chunk_size = df.extra_params.raw_chunk_size or options.chunk_size
    chunk_size = decide_dataframe_chunk_sizes(df.shape, chunk_size, memory_usage)
    chunk_size_idxes = (range(len(size)) for size in chunk_size)
    out_chunks = []
    index_values = dict()
    column_values = dict()
    for chunk_shape, chunk_idx in zip(
        itertools.product(*chunk_size), itertools.product(*chunk_size_idxes)
    ):
        chunk_op = op.copy().reset_key()
        slc = get_chunk_slices(chunk_size, chunk_idx)
        i_slc, j_slc = slc
        if j_slc == slice(0, df.shape[1]):
            # optimize full slice, it's way more faster
            j_slc = slice(None)
        chunk_op.data = raw_df.iloc[i_slc, j_slc]
        chunk_op.dtypes = chunk_op.data.dtypes
        i, j = chunk_idx
        if i in index_values:
            index_value = index_values[i]
        else:
            index_value = index_values[i] = parse_index(chunk_op.data.index)
        if j in column_values:
            column_value = column_values[j]
        else:
            column_value = column_values[j] = parse_index(
                chunk_op.data.columns, store_data=True
            )
        out_chunk = chunk_op.new_chunk(
            None,
            shape=chunk_shape,
            index=chunk_idx,
            index_value=index_value,
            columns_value=column_value,
            dtypes=chunk_op.data.dtypes,
        )
        out_chunks.append(out_chunk)
    new_op = op.copy()
    return new_op.new_dataframes(
        None,
        df.shape,
        dtypes=op.dtypes,
        index_value=df.index_value,
        columns_value=df.columns_value,
        chunks=out_chunks,
        nsplits=chunk_size,
        **df.extra_params
    )