arrow_avro/reader/async_reader/
store.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use crate::errors::AvroError;
19use crate::reader::async_reader::AsyncFileReader;
20use bytes::Bytes;
21use futures::future::BoxFuture;
22use futures::{FutureExt, TryFutureExt};
23use object_store::ObjectStore;
24use object_store::ObjectStoreExt;
25use object_store::path::Path;
26use std::error::Error;
27use std::ops::Range;
28use std::sync::Arc;
29use tokio::runtime::Handle;
30
31/// An implementation of an AsyncFileReader using the [`ObjectStore`] API.
32#[deprecated(
33    since = "59.2.0",
34    note = "Implement `AsyncFileReader` directly instead; see the example on the `AsyncFileReader` trait documentation and `arrow-avro/examples/object_store.rs`. Use `SpawnedReader` to perform I/O on a dedicated runtime. See also https://github.com/apache/arrow-rs/issues/10308"
35)]
36#[derive(Clone, Debug)]
37pub struct AvroObjectReader {
38    store: Arc<dyn ObjectStore>,
39    path: Path,
40    runtime: Option<Handle>,
41}
42
43#[allow(deprecated)]
44impl AvroObjectReader {
45    /// Creates a new [`Self`] from a store implementation and file location.
46    pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
47        Self {
48            store,
49            path,
50            runtime: None,
51        }
52    }
53
54    /// Perform IO on the provided tokio runtime
55    ///
56    /// Tokio is a cooperative scheduler, and relies on tasks yielding in a timely manner
57    /// to service IO. Therefore, running IO and CPU-bound tasks, such as avro decoding,
58    /// on the same tokio runtime can lead to degraded throughput, dropped connections and
59    /// other issues. For more information see [here].
60    ///
61    /// [here]: https://www.influxdata.com/blog/using-rustlangs-async-tokio-runtime-for-cpu-bound-tasks/
62    #[deprecated(
63        since = "59.2.0",
64        note = "Wrap the reader in a `SpawnedReader` instead, e.g. `SpawnedReader::new(reader, handle)`. See also https://github.com/apache/arrow-rs/issues/10308"
65    )]
66    pub fn with_runtime(self, handle: Handle) -> Self {
67        Self {
68            runtime: Some(handle),
69            ..self
70        }
71    }
72
73    fn spawn<F, O, E>(&self, f: F) -> BoxFuture<'_, Result<O, AvroError>>
74    where
75        F: for<'a> FnOnce(&'a Arc<dyn ObjectStore>, &'a Path) -> BoxFuture<'a, Result<O, E>>
76            + Send
77            + 'static,
78        O: Send + 'static,
79        E: Error + Send + 'static,
80    {
81        match &self.runtime {
82            Some(handle) => {
83                let path = self.path.clone();
84                let store = Arc::clone(&self.store);
85                handle
86                    .spawn(async move { f(&store, &path).await })
87                    .map_ok_or_else(
88                        |e| match e.try_into_panic() {
89                            Err(e) => Err(AvroError::External(Box::new(e))),
90                            Ok(p) => std::panic::resume_unwind(p),
91                        },
92                        |res| res.map_err(|e| AvroError::General(e.to_string())),
93                    )
94                    .boxed()
95            }
96            None => f(&self.store, &self.path)
97                .map_err(|e| AvroError::General(e.to_string()))
98                .boxed(),
99        }
100    }
101}
102
103#[allow(deprecated)]
104impl AsyncFileReader for AvroObjectReader {
105    fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes, AvroError>> {
106        self.spawn(|store, path| async move { store.get_range(path, range).await }.boxed())
107    }
108
109    fn get_byte_ranges(
110        &mut self,
111        ranges: Vec<Range<u64>>,
112    ) -> BoxFuture<'_, Result<Vec<Bytes>, AvroError>>
113    where
114        Self: Send,
115    {
116        self.spawn(|store, path| async move { store.get_ranges(path, &ranges).await }.boxed())
117    }
118}
arrow_avro/reader/async_reader/store.rs

arrow_avro/reader/async_reader/
store.rs
