parquet/arrow/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 std::{ops::Range, sync::Arc};
19
20use crate::arrow::arrow_reader::ArrowReaderOptions;
21use crate::arrow::async_reader::{AsyncFileReader, MetadataSuffixFetch};
22use crate::errors::{ParquetError, Result};
23use crate::file::metadata::{PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader};
24use bytes::Bytes;
25use futures::{FutureExt, TryFutureExt, future::BoxFuture};
26use object_store::ObjectStoreExt;
27use object_store::{GetOptions, GetRange};
28use object_store::{ObjectStore, path::Path};
29use tokio::runtime::Handle;
30/// Reads Parquet files in object storage using [`ObjectStore`].
31///
32/// ```no_run
33/// # use std::io::stdout;
34/// # use std::sync::Arc;
35/// # use object_store::azure::MicrosoftAzureBuilder;
36/// # use object_store::{ObjectStore, ObjectStoreExt};
37/// # use object_store::path::Path;
38/// # use parquet::arrow::async_reader::ParquetObjectReader;
39/// # use parquet::arrow::ParquetRecordBatchStreamBuilder;
40/// # use parquet::schema::printer::print_parquet_metadata;
41/// # async fn run() {
42/// // Populate configuration from environment
43/// let storage_container = Arc::new(MicrosoftAzureBuilder::from_env().build().unwrap());
44/// let location = Path::from("path/to/blob.parquet");
45/// let meta = storage_container.head(&location).await.unwrap();
46/// println!("Found Blob with {}B at {}", meta.size, meta.location);
47///
48/// // Show Parquet metadata
49/// let reader = ParquetObjectReader::new(storage_container, meta.location).with_file_size(meta.size);
50/// let builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap();
51/// print_parquet_metadata(&mut stdout(), builder.metadata());
52/// # }
53/// ```
54#[deprecated(
55    since = "59.2.0",
56    note = "Implement `AsyncFileReader` directly instead; see the example on the `AsyncFileReader` trait documentation and `parquet/examples/object_store.rs`. Use `SpawnedReader` to perform I/O on a dedicated runtime. See https://github.com/apache/arrow-rs/issues/10308"
57)]
58#[derive(Clone, Debug)]
59pub struct ParquetObjectReader {
60    store: Arc<dyn ObjectStore>,
61    path: Path,
62    file_size: Option<u64>,
63    metadata_size_hint: Option<usize>,
64    preload_column_index: bool,
65    preload_offset_index: bool,
66    runtime: Option<Handle>,
67}
68
69#[expect(deprecated)]
70impl ParquetObjectReader {
71    /// Creates a new [`ParquetObjectReader`] for the provided [`ObjectStore`] and [`Path`].
72    pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
73        Self {
74            store,
75            path,
76            file_size: None,
77            metadata_size_hint: None,
78            preload_column_index: false,
79            preload_offset_index: false,
80            runtime: None,
81        }
82    }
83
84    /// Provide a hint as to the size of the parquet file's footer,
85    /// see [`ParquetMetaDataReader::with_prefetch_hint`]
86    pub fn with_footer_size_hint(self, hint: usize) -> Self {
87        Self {
88            metadata_size_hint: Some(hint),
89            ..self
90        }
91    }
92
93    /// Provide the byte size of this file.
94    ///
95    /// If provided, the file size will ensure that only bounded range requests are used. If file
96    /// size is not provided, the reader will use suffix range requests to fetch the metadata.
97    ///
98    /// Providing this size up front is an important optimization to avoid extra calls when the
99    /// underlying store does not support suffix range requests.
100    ///
101    /// The file size can be obtained using [`ObjectStore::list`] or [`ObjectStoreExt::head`].
102    pub fn with_file_size(self, file_size: u64) -> Self {
103        Self {
104            file_size: Some(file_size),
105            ..self
106        }
107    }
108
109    /// Whether to load the Column Index as part of [`Self::get_metadata`]
110    ///
111    /// Note: This setting may be overridden by [`ArrowReaderOptions`] `page_index_policy`.
112    /// If `page_index_policy` is `Optional` or `Required`, it will take precedence
113    /// over this preload flag. When it is `Skip` (default), this flag is used.
114    pub fn with_preload_column_index(self, preload_column_index: bool) -> Self {
115        Self {
116            preload_column_index,
117            ..self
118        }
119    }
120
121    /// Whether to load the Offset Index as part of [`Self::get_metadata`]
122    ///
123    /// Note: This setting may be overridden by [`ArrowReaderOptions`] `page_index_policy`.
124    /// If `page_index_policy` is `Optional` or `Required`, it will take precedence
125    /// over this preload flag. When it is `Skip` (default), this flag is used.
126    pub fn with_preload_offset_index(self, preload_offset_index: bool) -> Self {
127        Self {
128            preload_offset_index,
129            ..self
130        }
131    }
132
133    /// Perform IO on the provided tokio runtime
134    ///
135    /// Tokio is a cooperative scheduler, and relies on tasks yielding in a timely manner
136    /// to service IO. Therefore, running IO and CPU-bound tasks, such as parquet decoding,
137    /// on the same tokio runtime can lead to degraded throughput, dropped connections and
138    /// other issues. For more information see [here].
139    ///
140    /// [here]: https://www.influxdata.com/blog/using-rustlangs-async-tokio-runtime-for-cpu-bound-tasks/
141    #[deprecated(
142        since = "59.2.0",
143        note = "Wrap the reader in a `SpawnedReader` instead, e.g. `SpawnedReader::new(reader, handle)`. See https://github.com/apache/arrow-rs/issues/10308"
144    )]
145    pub fn with_runtime(self, handle: Handle) -> Self {
146        Self {
147            runtime: Some(handle),
148            ..self
149        }
150    }
151
152    fn spawn<F, O, E>(&self, f: F) -> BoxFuture<'_, Result<O>>
153    where
154        F: for<'a> FnOnce(&'a Arc<dyn ObjectStore>, &'a Path) -> BoxFuture<'a, Result<O, E>>
155            + Send
156            + 'static,
157        O: Send + 'static,
158        E: Into<ParquetError> + Send + 'static,
159    {
160        match &self.runtime {
161            Some(handle) => {
162                let path = self.path.clone();
163                let store = Arc::clone(&self.store);
164                handle
165                    .spawn(async move { f(&store, &path).await })
166                    .map_ok_or_else(
167                        |e| match e.try_into_panic() {
168                            Err(e) => Err(ParquetError::External(Box::new(e))),
169                            Ok(p) => std::panic::resume_unwind(p),
170                        },
171                        |res| res.map_err(|e| e.into()),
172                    )
173                    .boxed()
174            }
175            None => f(&self.store, &self.path).map_err(|e| e.into()).boxed(),
176        }
177    }
178}
179
180#[expect(deprecated)]
181impl MetadataSuffixFetch for &mut ParquetObjectReader {
182    fn fetch_suffix(&mut self, suffix: usize) -> BoxFuture<'_, Result<Bytes>> {
183        let options = GetOptions {
184            range: Some(GetRange::Suffix(suffix as u64)),
185            ..Default::default()
186        };
187        self.spawn(|store, path| {
188            async move {
189                let resp = store.get_opts(path, options).await?;
190                Ok::<_, ParquetError>(resp.bytes().await?)
191            }
192            .boxed()
193        })
194    }
195}
196
197#[expect(deprecated)]
198impl AsyncFileReader for ParquetObjectReader {
199    fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes>> {
200        self.spawn(|store, path| store.get_range(path, range).boxed())
201    }
202
203    fn get_byte_ranges(&mut self, ranges: Vec<Range<u64>>) -> BoxFuture<'_, Result<Vec<Bytes>>>
204    where
205        Self: Send,
206    {
207        self.spawn(|store, path| async move { store.get_ranges(path, &ranges).await }.boxed())
208    }
209
210    // This method doesn't directly call `self.spawn` because all of the IO that is done down the
211    // line due to this method call is done through `self.get_bytes` and/or `self.get_byte_ranges`.
212    // When `self` is passed into `ParquetMetaDataReader::load_and_finish`, it treats it as
213    // an `impl MetadataFetch` and calls those methods to get data from it. Due to `Self`'s impl of
214    // `AsyncFileReader`, the calls to `MetadataFetch::fetch` are just delegated to
215    // `Self::get_bytes`.
216    fn get_metadata<'a>(
217        &'a mut self,
218        options: Option<&'a ArrowReaderOptions>,
219    ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
220        Box::pin(async move {
221            let metadata_opts = options.map(|o| o.metadata_options().clone());
222            let mut metadata = ParquetMetaDataReader::new()
223                .with_metadata_options(metadata_opts)
224                .with_column_index_policy(PageIndexPolicy::from(self.preload_column_index))
225                .with_offset_index_policy(PageIndexPolicy::from(self.preload_offset_index))
226                .with_prefetch_hint(self.metadata_size_hint);
227
228            #[cfg(feature = "encryption")]
229            if let Some(options) = options {
230                metadata = metadata.with_decryption_properties(
231                    options.file_decryption_properties.as_ref().map(Arc::clone),
232                );
233            }
234
235            // Override page index policies from ArrowReaderOptions if specified and not Skip.
236            // When page_index_policy is Skip (default), use the reader's preload flags.
237            // When page_index_policy is Optional or Required, override the preload flags
238            // to ensure the specified policy takes precedence.
239            if let Some(options) = options
240                && (options.column_index_policy() != PageIndexPolicy::Skip
241                    || options.offset_index_policy() != PageIndexPolicy::Skip)
242            {
243                metadata = metadata
244                    .with_column_index_policy(options.column_index_policy())
245                    .with_offset_index_policy(options.offset_index_policy());
246            }
247            // TODO: https://github.com/apache/arrow-rs/issues/11395 may deprecate this API. Decide
248            // at that time if masks should be added to ParquetObjectReader.
249            if let Some(options) = options {
250                if !options.column_index_mask().is_all() {
251                    metadata = metadata.with_column_index_mask(options.column_index_mask().clone());
252                }
253                if !options.offset_index_mask().is_all() {
254                    metadata = metadata.with_offset_index_mask(options.offset_index_mask().clone());
255                }
256            }
257
258            let metadata = if let Some(file_size) = self.file_size {
259                metadata.load_and_finish(self, file_size).await?
260            } else {
261                metadata.load_via_suffix_and_finish(self).await?
262            };
263
264            Ok(Arc::new(metadata))
265        })
266    }
267}
268
269#[cfg(test)]
270#[expect(deprecated)]
271mod tests {
272    use crate::arrow::async_reader::ArrowReaderOptions;
273    use crate::file::metadata::PageIndexPolicy;
274    use std::sync::{
275        Arc,
276        atomic::{AtomicUsize, Ordering},
277    };
278
279    use futures::TryStreamExt;
280
281    use crate::arrow::ParquetRecordBatchStreamBuilder;
282    use crate::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
283    use crate::errors::ParquetError;
284    use arrow::util::test_util::parquet_test_data;
285    use futures::FutureExt;
286    use object_store::local::LocalFileSystem;
287    use object_store::path::Path;
288    use object_store::{ObjectMeta, ObjectStore, ObjectStoreExt};
289
290    async fn get_meta_store() -> (ObjectMeta, Arc<dyn ObjectStore>) {
291        let res = parquet_test_data();
292        let store = LocalFileSystem::new_with_prefix(res).unwrap();
293
294        let meta = store
295            .head(&Path::from("alltypes_plain.parquet"))
296            .await
297            .unwrap();
298
299        (meta, Arc::new(store) as Arc<dyn ObjectStore>)
300    }
301
302    async fn get_meta_store_with_page_index() -> (ObjectMeta, Arc<dyn ObjectStore>) {
303        let res = parquet_test_data();
304        let store = LocalFileSystem::new_with_prefix(res).unwrap();
305
306        let meta = store
307            .head(&Path::from("alltypes_tiny_pages_plain.parquet"))
308            .await
309            .unwrap();
310
311        (meta, Arc::new(store) as Arc<dyn ObjectStore>)
312    }
313
314    #[tokio::test]
315    async fn test_simple() {
316        let (meta, store) = get_meta_store().await;
317        let object_reader =
318            ParquetObjectReader::new(store, meta.location).with_file_size(meta.size);
319
320        let builder = ParquetRecordBatchStreamBuilder::new(object_reader)
321            .await
322            .unwrap();
323        let batches: Vec<_> = builder.build().unwrap().try_collect().await.unwrap();
324
325        assert_eq!(batches.len(), 1);
326        assert_eq!(batches[0].num_rows(), 8);
327    }
328
329    #[tokio::test]
330    async fn test_simple_without_file_length() {
331        let (meta, store) = get_meta_store().await;
332        let object_reader = ParquetObjectReader::new(store, meta.location);
333
334        let builder = ParquetRecordBatchStreamBuilder::new(object_reader)
335            .await
336            .unwrap();
337        let batches: Vec<_> = builder.build().unwrap().try_collect().await.unwrap();
338
339        assert_eq!(batches.len(), 1);
340        assert_eq!(batches[0].num_rows(), 8);
341    }
342
343    #[tokio::test]
344    async fn test_not_found() {
345        let (mut meta, store) = get_meta_store().await;
346        meta.location = Path::from("I don't exist.parquet");
347
348        let object_reader =
349            ParquetObjectReader::new(store, meta.location).with_file_size(meta.size);
350        // Cannot use unwrap_err as ParquetRecordBatchStreamBuilder: !Debug
351        match ParquetRecordBatchStreamBuilder::new(object_reader).await {
352            Ok(_) => panic!("expected failure"),
353            Err(e) => {
354                let err = e.to_string();
355                assert!(err.contains("I don't exist.parquet not found:"), "{err}");
356            }
357        }
358    }
359
360    #[tokio::test]
361    async fn test_runtime_is_used() {
362        let num_actions = Arc::new(AtomicUsize::new(0));
363
364        let (a1, a2) = (num_actions.clone(), num_actions.clone());
365        let rt = tokio::runtime::Builder::new_multi_thread()
366            .on_thread_park(move || {
367                a1.fetch_add(1, Ordering::Relaxed);
368            })
369            .on_thread_unpark(move || {
370                a2.fetch_add(1, Ordering::Relaxed);
371            })
372            .build()
373            .unwrap();
374
375        let (meta, store) = get_meta_store().await;
376
377        let initial_actions = num_actions.load(Ordering::Relaxed);
378
379        let reader = ParquetObjectReader::new(store, meta.location)
380            .with_file_size(meta.size)
381            .with_runtime(rt.handle().clone());
382
383        let builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap();
384        let batches: Vec<_> = builder.build().unwrap().try_collect().await.unwrap();
385
386        // Just copied these assert_eqs from the `test_simple` above
387        assert_eq!(batches.len(), 1);
388        assert_eq!(batches[0].num_rows(), 8);
389
390        assert!(num_actions.load(Ordering::Relaxed) - initial_actions > 0);
391
392        // Runtimes have to be dropped in blocking contexts, so we need to move this one to a new
393        // blocking thread to drop it.
394        tokio::runtime::Handle::current().spawn_blocking(move || drop(rt));
395    }
396
397    /// Unit test that `ParquetObjectReader::spawn`spawns on the provided runtime
398    #[tokio::test]
399    async fn test_runtime_thread_id_different() {
400        let rt = tokio::runtime::Builder::new_multi_thread()
401            .worker_threads(1)
402            .build()
403            .unwrap();
404
405        let (meta, store) = get_meta_store().await;
406
407        let reader = ParquetObjectReader::new(store, meta.location)
408            .with_file_size(meta.size)
409            .with_runtime(rt.handle().clone());
410
411        let current_id = std::thread::current().id();
412
413        let other_id = reader
414            .spawn(|_, _| async move { Ok::<_, ParquetError>(std::thread::current().id()) }.boxed())
415            .await
416            .unwrap();
417
418        assert_ne!(current_id, other_id);
419
420        tokio::runtime::Handle::current().spawn_blocking(move || drop(rt));
421    }
422
423    #[tokio::test]
424    async fn io_fails_on_shutdown_runtime() {
425        let rt = tokio::runtime::Builder::new_multi_thread()
426            .worker_threads(1)
427            .build()
428            .unwrap();
429
430        let (meta, store) = get_meta_store().await;
431
432        let mut reader = ParquetObjectReader::new(store, meta.location)
433            .with_file_size(meta.size)
434            .with_runtime(rt.handle().clone());
435
436        rt.shutdown_background();
437
438        let err = reader.get_bytes(0..1).await.unwrap_err().to_string();
439
440        assert!(err.contains("was cancelled"));
441    }
442
443    #[tokio::test]
444    async fn test_page_index_policy_skip_uses_preload_true() {
445        let (meta, store) = get_meta_store_with_page_index().await;
446
447        // Create reader with preload flags set to true
448        let mut reader = ParquetObjectReader::new(store.clone(), meta.location.clone())
449            .with_file_size(meta.size)
450            .with_preload_column_index(true)
451            .with_preload_offset_index(true);
452
453        // Create options with page_index_policy set to Skip (default)
454        let options = ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Skip);
455
456        // Get metadata - Skip means use reader's preload flags (true)
457        let metadata = reader.get_metadata(Some(&options)).await.unwrap();
458
459        // With preload=true, indexes should be loaded since the test file has them
460        assert!(metadata.page_index().is_some_and(|idx| idx.is_complete()));
461    }
462
463    #[tokio::test]
464    async fn test_page_index_policy_optional_overrides_preload_false() {
465        let (meta, store) = get_meta_store_with_page_index().await;
466
467        // Create reader with preload flags set to false
468        let mut reader = ParquetObjectReader::new(store.clone(), meta.location.clone())
469            .with_file_size(meta.size)
470            .with_preload_column_index(false)
471            .with_preload_offset_index(false);
472
473        // Create options with page_index_policy set to Optional
474        let options = ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Optional);
475
476        // Get metadata - Optional overrides preload flags and attempts to load indexes
477        let metadata = reader.get_metadata(Some(&options)).await.unwrap();
478
479        // With Optional policy, it will TRY to load indexes but won't fail if they don't exist
480        // The test file has page indexes, so they will be some
481        assert!(metadata.page_index().is_some());
482    }
483
484    #[tokio::test]
485    async fn test_page_index_policy_optional_vs_skip() {
486        let (meta, store) = get_meta_store_with_page_index().await;
487
488        // Test 1: preload=false + Skip policy -> uses preload flags (false)
489        let mut reader1 = ParquetObjectReader::new(store.clone(), meta.location.clone())
490            .with_file_size(meta.size)
491            .with_preload_column_index(false)
492            .with_preload_offset_index(false);
493
494        let options1 = ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Skip);
495        let metadata1 = reader1.get_metadata(Some(&options1)).await.unwrap();
496
497        // Test 2: preload=false + Optional policy -> overrides to try loading
498        let mut reader2 = ParquetObjectReader::new(store.clone(), meta.location.clone())
499            .with_file_size(meta.size)
500            .with_preload_column_index(false)
501            .with_preload_offset_index(false);
502
503        let options2 = ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Optional);
504        let metadata2 = reader2.get_metadata(Some(&options2)).await.unwrap();
505
506        // Both should succeed (no panic/error)
507        // metadata1 (Skip) uses preload=false -> Skip policy
508        // metadata2 (Optional) overrides preload=false -> Optional policy
509        assert!(metadata1.page_index().is_none());
510        assert!(metadata2.page_index().is_some());
511    }
512
513    #[tokio::test]
514    async fn test_page_index_policy_no_options_uses_preload() {
515        let (meta, store) = get_meta_store_with_page_index().await;
516
517        // Create reader with preload flags set to true
518        let mut reader = ParquetObjectReader::new(store, meta.location)
519            .with_file_size(meta.size)
520            .with_preload_column_index(true)
521            .with_preload_offset_index(true);
522
523        // Get metadata without options - should use reader's preload flags
524        let metadata = reader.get_metadata(None).await.unwrap();
525
526        // With no options provided, preload flags (true) should be respected
527        // and converted to Optional policy internally (preload=true -> Optional)
528        // The test file has page indexes, so they will be some
529        assert!(metadata.page_index().is_some_and(|idx| idx.is_complete()));
530    }
531}
parquet/arrow/async_reader/store.rs

parquet/arrow/async_reader/
store.rs
