(rust/ python) Copy data from one operator to another

[chitralverma]

For a file, glob pattern or directory on one operator (local fs) i want to copy data (recursively?) to remote object store (s3, gcs ...) in the most efficient (preferably in zero-copy or near zero-copy) manner.

I wanted to do this in python with the efficient chunked reader and writer but those APIs are not are not available and open must be used. This is a point of improvement for python bindings.

In rust, the APIs are available but the Reader and Writer must be converted to FuturesAsyncReader and FuturesAsyncWriter respectively for a streaming or chunked write to destination. This gets a little confusing because into_futures_async_read requires a range bound but then a chunk_size was also defined on the reader as well. Besides, if i'm using tokio for copy operations then compat also needs to be added.

Question: So is there a way to improve these?

Python snippet

from opendal import AsyncOperator

async def copy_file(
    key,
    src_link: AsyncOperator,
    dest_link: AsyncOperator,
    chunk_size: int = 8 * 1024 * 1024, # 8MB buffer
):
    """Copies a single file asynchronously."""
    try:
        async with (
            await src_link.open(key, "rb") as reader,
            await dest_link.open(key, "wb") as writer,
        ):
            while True:
                chunk = await reader.read(chunk_size)
                if not chunk:
                    break
                await writer.write(chunk)

        logger.info(f"File '{key}' successfully copied")
        return True
    except Exception as e:
        logger.error(f"Error copying '{key}': {e}")
        return False

src_link = ???
dest_link = ???
key = "/path/to/file"

# Using this with asyncio
await asyncio.gather(*[copy_file(key, src_link, dest_link)])

Rust snippet

use opendal::{Operator, Reader, Writer};
use tokio::io::AsyncWriteExt;
use tokio_util::compat::{FuturesAsyncReadCompatExt, FuturesAsyncWriteCompatExt};
use anyhow::Result;

#[tokio::main]
async fn main() -> Result<()> {
    let BUFFER_SIZE: usize = 8 * 1024 * 1024; // 8MB buffer
    let THREADS: usize = ???
    
    let src_link: Operator = ???;
    let dest_link: Operator = ???;
    let key = "/path/to/file";
    
    let mut reader = src_link
        .reader_with(key).chunk(BUFFER_SIZE).concurrent(THREADS)
        .await?
        .into_futures_async_read(..10) // fails if size of the file is unknown, example: this will fail if file is of size less than 10 bytes
        .await?
        .compat();
    
    let mut writer = dest_link
        .writer_with(key).chunk(BUFFER_SIZE).concurrent(THREADS) // should chunks and concurrent be set on writer as well?
        .await?
        .into_futures_async_write()
        .compat_write();
    
    let num_bytes = tokio::io::copy(&mut reader, &mut writer).await?;
    println!("File '{key}' successfully copied with {num_bytes} bytes!");
    
    writer.flush().await?;
    Ok(())
}

[chitralverma]

CC @Xuanwo

[Xuanwo]

For the Rust part, I think we can expose BufferStream and BufferSink to allow direct reuse of the underlying Buffer: #4592. I believe all problems raises in rust code can be solved by this.

For the Python part, I'm not sure what "the efficient chunked reader and writer" refers to. Are you talking about OpenDAL's chunked Buffer?

[Xuanwo]

Ok, let's add BufferStream and BufferSink first, I will take a look.

Also invite python maintainers to take a look on python's API @messense @Zheaoli @PsiACE

[chitralverma]

@Xuanwo whats the significance of RangeBounds in into_futures_async_read(<RangeBounds>)?
if i'm giving an unbounded range .. with chunk size let's say 8MB and some value for concurrent on the reader, how does this work ?

unfortunately there are no docs regarding this

[Xuanwo]

Hi, the returning struct from into_futures_async_read implements both AsyncRead and AsyncSeek so it need to know the exact size of this file. If the size is not unknown before, opendal will call stat to get the size when users call seek.

If users pass .. and set a chunk, it will stat this file first to get the total size so that we know how many chunks we need.

I will add this in docs.

[Xuanwo]

Ok, let's add BufferStream and BufferSink first, I will take a look.

This API will be available after #5698

[chitralverma]

@messense @Zheaoli @PsiACE can we expose this read and writer API with chunking and concurrent in python binding as well?

[chitralverma]

See comment