Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

attempt at list stream #13

Closed
wants to merge 4 commits into from
Closed

attempt at list stream #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e0faacc
attempt at list stream
kylebarron Oct 18, 2024
39d086a
Merge branch 'main' into kyle/list-stream
kylebarron Oct 18, 2024
b86ae9d
Merge branch 'main' into kyle/list-stream
kylebarron Oct 22, 2024
2f02ca5
Attempt ouroboros
kylebarron Oct 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 75 additions & 3 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions object-store-rs/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ futures = { workspace = true }
http = { workspace = true }
indexmap = { workspace = true }
object_store = { workspace = true }
ouroboros = "0.18"
pyo3 = { workspace = true, features = ["chrono", "abi3-py39"] }
pyo3-async-runtimes = { workspace = true, features = ["tokio-runtime"] }
pyo3-file = { workspace = true }
Expand Down
138 changes: 133 additions & 5 deletions object-store-rs/src/list.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,12 @@ use futures::StreamExt;
use indexmap::IndexMap;
use object_store::path::Path;
use object_store::{ListResult, ObjectMeta, ObjectStore};
use ouroboros::self_referencing;
use pyo3::exceptions::{PyStopAsyncIteration, PyStopIteration};
use pyo3::prelude::*;
use pyo3_object_store::error::{PyObjectStoreError, PyObjectStoreResult};
use pyo3_object_store::PyObjectStore;
use tokio::sync::Mutex;

use crate::runtime::get_runtime;

Expand All @@ -33,9 +36,9 @@ impl IntoPy<PyObject> for PyObjectMeta {
}
}

pub(crate) struct PyListResult(ListResult);
pub(crate) struct PyMaterializedListResult(ListResult);

impl IntoPy<PyObject> for PyListResult {
impl IntoPy<PyObject> for PyMaterializedListResult {
fn into_py(self, py: Python<'_>) -> PyObject {
let mut dict = IndexMap::with_capacity(2);
dict.insert(
Expand All @@ -60,6 +63,131 @@ impl IntoPy<PyObject> for PyListResult {
}
}

#[pyclass(name = "ListResult")]
#[self_referencing]
pub(crate) struct PyListResult {
store: Arc<dyn ObjectStore>,
#[borrows(store)]
payload: BoxStream<'this, object_store::Result<ObjectMeta>>,
}

// impl PyListResult {
// fn new(
// // store: Arc<dyn ObjectStore>,
// // prefix: Option<&Path>,
// stream: BoxStream<'static, object_store::Result<ObjectMeta>>,
// min_chunk_size: usize,
// ) -> Self {
// // let stream = store.as_ref().list(prefix);
// Self {
// payload: stream,
// min_chunk_size,
// }
// }
// }

// #[pyclass(name = "ListStream")]
// pub struct PyListStream {
// stream: Arc<Mutex<BoxStream<'static, object_store::Result<ObjectMeta>>>>,
// min_chunk_size: usize,
// }

// impl PyListStream {
// fn new(
// stream: BoxStream<'static, object_store::Result<ObjectMeta>>,
// min_chunk_size: usize,
// ) -> Self {
// Self {
// stream: Arc::new(Mutex::new(stream)),
// min_chunk_size,
// }
// }
// }

// async fn next_stream(
// stream: Arc<Mutex<BoxStream<'static, object_store::Result<ObjectMeta>>>>,
// min_chunk_size: usize,
// sync: bool,
// ) -> PyResult<()> {
// let mut stream = stream.lock().await;
// let mut metas: Vec<ObjectMeta> = vec![];
// loop {
// match stream.next().await {
// Some(Ok(meta)) => {
// metas.push(meta);
// if metas.len() >= min_chunk_size {
// todo!()
// // return Ok(PyBytesWrapper::new_multiple(buffers));
// }
// }
// Some(Err(e)) => return Err(PyObjectStoreError::from(e).into()),
// None => {
// if metas.is_empty() {
// // Depending on whether the iteration is sync or not, we raise either a
// // StopIteration or a StopAsyncIteration
// if sync {
// return Err(PyStopIteration::new_err("stream exhausted"));
// } else {
// return Err(PyStopAsyncIteration::new_err("stream exhausted"));
// }
// } else {
// todo!()
// // return Ok(PyBytesWrapper::new_multiple(buffers));
// }
// }
// };
// }
// }

// #[pymethods]
// impl PyListStream {
// fn __aiter__(slf: Py<Self>) -> Py<Self> {
// slf
// }

// fn __iter__(slf: Py<Self>) -> Py<Self> {
// slf
// }

// fn __anext__<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<PyAny>> {
// let stream = self.stream.clone();
// pyo3_async_runtimes::tokio::future_into_py(
// py,
// next_stream(stream, self.min_chunk_size, false),
// )
// }

// fn __next__<'py>(&'py self, py: Python<'py>) -> PyResult<()> {
// let runtime = get_runtime(py)?;
// let stream = self.stream.clone();
// runtime.block_on(next_stream(stream, self.min_chunk_size, true))
// }
// }

// #[pyo3(signature = (store, prefix = None, *, min_chunk_size = 1000))]
// pub(crate) fn list<'py>(
// py: Python<'py>,
// store: PyObjectStore,
// prefix: Option<String>,
// min_chunk_size: usize,
// ) -> PyObjectStoreResult<PyListResult> {
// // todo!()
// // // let runtime = get_runtime(py)?;
// // let store = store.into_inner();
// let stream = store.as_ref().list(prefix.map(|s| s.into()).as_ref());
// // todo!()
// Ok(PyListResult::new(stream, min_chunk_size))
// // Ok::<_, PyObjectStoreError>(PyListResult::new(store, stream, min_chunk_size))
// // // py.allow_threads(move || {

// // // let x = runtime.block_on(fut);
// // // let out = runtime.block_on(list_materialize(
// // // store.into_inner(),
// // // prefix.map(|s| s.into()).as_ref(),
// // // ))?;
// // // Ok::<_, PyObjectStoreError>(out)
// // })

#[pyfunction]
#[pyo3(signature = (store, prefix = None, *, offset = None, max_items = 2000))]
pub(crate) fn list(
Expand Down Expand Up @@ -128,7 +256,7 @@ pub(crate) fn list_with_delimiter(
py: Python,
store: PyObjectStore,
prefix: Option<String>,
) -> PyObjectStoreResult<PyListResult> {
) -> PyObjectStoreResult<PyMaterializedListResult> {
let runtime = get_runtime(py)?;
py.allow_threads(|| {
let out = runtime.block_on(list_with_delimiter_materialize(
Expand Down Expand Up @@ -157,7 +285,7 @@ pub(crate) fn list_with_delimiter_async(
async fn list_with_delimiter_materialize(
store: Arc<dyn ObjectStore>,
prefix: Option<&Path>,
) -> PyObjectStoreResult<PyListResult> {
) -> PyObjectStoreResult<PyMaterializedListResult> {
let list_result = store.list_with_delimiter(prefix).await?;
Ok(PyListResult(list_result))
Ok(PyMaterializedListResult(list_result))
}
Loading