def METHOD_NAME(  # noqa: C901, pylint: disable=too-many-statements
    vineyard_socket: str,
    path: str,
    storage_options: Dict,
    read_options: Dict,
    proc_num: int,
    proc_index: int,
):
    """Read bytes from external storage and produce a ByteStream,
    which will later be assembled into a ParallelStream.
    Args:
        vineyard_socket (str): Ipc socket
        path (str): External storage path to write to
        storage_options (dict): Configurations of external storage
        read_options (dict): Additional options that could control the
                             behavior of read
        proc_num (int): Total amount of process
        proc_index (int): The sequence of this process
    Raises:
        ValueError: If the stream is invalid.
    """
    client = vineyard.connect(vineyard_socket)
    params = dict()
    # Cut strings after # and read potential kvs
    parts = path.split('#', 1)
    path = parts[0]
    if len(parts) > 1:
        options = parts[1]
        for split_by_hashtap in options.split('#'):
            for split_by_ampersand in split_by_hashtap.split('&'):
                k, v = split_by_ampersand.split('=')
                read_options[k] = v
    read_block_delimiter = read_options.pop('read_block_delimiter', '\n')
    if read_block_delimiter is not None:
        read_block_delimiter = read_block_delimiter.encode('utf-8')
    # Used when reading tables from external storage.
    # Usually for load a property graph
    # header_row: each file has a header_row when reading from directories
    header_row = bool(read_options.get("header_row", False))
    # first_header_row: only the first file (alphabetical ordered) has a header_row
    # when reading from directories
    first_header_row = read_options.get("first_header_row", False)
    for k, v in read_options.items():
        if k in ("header_row", "first_header_row", "include_all_columns", "accumulate"):
            params[k] = "1" if v else "0"
        elif k == "delimiter":
            params[k] = bytes(v, "utf-8").decode("unicode_escape")
        elif not isinstance(v, str):
            params[k] = repr(v)
        else:
            params[k] = v
    try:
        # files would be empty if it's a glob pattern and globbed nothing.
        fs, _, files = get_fs_token_paths(path, storage_options=storage_options)
    except Exception:  # pylint: disable=broad-except
        report_error(
            f"Cannot initialize such filesystem for '{path}', "
            f"exception is:\n{traceback.format_exc()}"
        )
        sys.exit(-1)
    try:
        assert files
    except Exception:  # pylint: disable=broad-except
        report_error(f"Cannot find such files for '{path}'")
        sys.exit(-1)
    files = sorted(files)
    stream, writer = None, None
    if 'chunk_size' in storage_options:
        chunk_size = parse_readable_size(storage_options['chunk_size'])
    else:
        chunk_size = 1024 * 1024 * 128  # default: 64MB
    try:
        for index, file_path in enumerate(files):
            with fs.open(file_path, mode="rb") as fp:
                offset = 0
                # Only process header line when processing first file
                # And open the writer when processing first file
                if header_row or (first_header_row and index == 0):
                    header_line = read_block(fp, 0, 1, read_block_delimiter)
                    params["header_line"] = header_line.decode("unicode_escape")
                    offset = len(header_line)
                    # strip the UTF-8 BOM
                    if header_line[0:3] == b'\xef\xbb\xbf':
                        header_line = header_line[3:]
                if index == 0:
                    stream = ByteStream.new(client, params)
                    client.persist(stream.id)
                    report_success(stream.id)
                    writer = stream.open_writer(client)
                read_byte_blocks(
                    fp,
                    proc_num,
                    proc_index,
                    index,
                    header_row,
                    offset,
                    chunk_size,
                    read_block_delimiter,
                    writer,
                )
        writer.finish()
    except Exception:  # pylint: disable=broad-except
        report_exception()
        if writer is not None:
            writer.fail()
        sys.exit(-1)