if isinstance(self.path_or_buf, (str, bytes, os.PathLike)):
            with fsspec.open(self.path_or_buf, "wb", **(self.storage_options or {})) as buffer:
                written = self._write(file_obj=buffer, header=header, index=index, **self.to_csv_kwargs)
        else:
            written = self._write(file_obj=self.path_or_buf, header=header, index=index, **self.to_csv_kwargs)
        return written

    def _batch_csv(self, args):
        offset, header, index, to_csv_kwargs = args

        batch = query_table(
            table=self.dataset.data,
            key=slice(offset, offset + self.batch_size),
            indices=self.dataset._indices,
        )
        csv_str = batch.to_pandas().to_csv(
            path_or_buf=None, header=header if (offset == 0) else False, index=index, **to_csv_kwargs
        )
        return csv_str.encode(self.encoding)