-
Notifications
You must be signed in to change notification settings - Fork 46
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: use parquet lru reader as trait object
- Loading branch information
Showing
19 changed files
with
263 additions
and
309 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| use std::{ops::Range, sync::Arc}; | ||
|
|
||
| use bytes::Bytes; | ||
| use futures_core::future::BoxFuture; | ||
| use parquet::{ | ||
| arrow::async_reader::AsyncFileReader, errors::Result, file::metadata::ParquetMetaData, | ||
| }; | ||
|
|
||
| use crate::LruCache; | ||
|
|
||
| pub struct BoxedFileReader { | ||
| inner: Box<dyn AsyncFileReader>, | ||
| } | ||
|
|
||
| impl BoxedFileReader { | ||
| pub fn new<T: AsyncFileReader + 'static>(inner: T) -> Self { | ||
| Self { | ||
| inner: Box::new(inner), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl AsyncFileReader for BoxedFileReader { | ||
| fn get_bytes(&mut self, range: Range<usize>) -> BoxFuture<'_, Result<Bytes>> { | ||
| self.inner.get_bytes(range) | ||
| } | ||
|
|
||
| fn get_metadata(&mut self) -> BoxFuture<'_, Result<Arc<ParquetMetaData>>> { | ||
| self.inner.get_metadata() | ||
| } | ||
|
|
||
| fn get_byte_ranges(&mut self, ranges: Vec<Range<usize>>) -> BoxFuture<'_, Result<Vec<Bytes>>> { | ||
| self.inner.get_byte_ranges(ranges) | ||
| } | ||
| } | ||
|
|
||
| pub trait DynLruCache<K> { | ||
| fn get_reader(&self, key: K, reader: BoxedFileReader) -> BoxFuture<'_, BoxedFileReader>; | ||
| } | ||
|
|
||
| impl<K, C> DynLruCache<K> for C | ||
| where | ||
| K: 'static + Send, | ||
| C: LruCache<K> + Sized + Send + Sync, | ||
| { | ||
| fn get_reader(&self, key: K, reader: BoxedFileReader) -> BoxFuture<'_, BoxedFileReader> { | ||
| Box::pin(async move { BoxedFileReader::new(self.get_reader(key, reader).await) }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,63 +1,56 @@ | ||
| mod r#dyn; | ||
| #[cfg(feature = "foyer")] | ||
| pub mod foyer; | ||
|
|
||
| use std::{future::Future, marker::PhantomData}; | ||
|
|
||
| use parquet::{arrow::async_reader::AsyncFileReader, errors::Result}; | ||
| use thiserror::Error; | ||
| use parquet::arrow::async_reader::AsyncFileReader; | ||
|
|
||
| #[derive(Default)] | ||
| pub struct Options { | ||
| meta_capacity: usize, | ||
| data_capacity: usize, | ||
| } | ||
|
|
||
| impl Options { | ||
| pub fn meta_capacity(mut self, meta_capacity: usize) -> Self { | ||
| self.meta_capacity = meta_capacity; | ||
| self | ||
| } | ||
|
|
||
| pub fn data_capacity(mut self, data_capacity: usize) -> Self { | ||
| self.data_capacity = data_capacity; | ||
| self | ||
| } | ||
| } | ||
| pub use crate::r#dyn::*; | ||
|
|
||
| pub trait LruCache<K>: Clone + Send + Sync + 'static { | ||
| type LruReader<R: AsyncFileReader + 'static>: AsyncFileReader + 'static; | ||
|
|
||
| fn new(options: Options) -> impl Future<Output = Result<Self, Error>> + Send; | ||
| pub trait LruCache<K> | ||
| where | ||
| K: 'static, | ||
| { | ||
| type LruReader<R>: AsyncFileReader + 'static | ||
| where | ||
| R: AsyncFileReader + 'static; | ||
|
|
||
| fn get_reader<R>(&self, key: K, reader: R) -> impl Future<Output = Self::LruReader<R>> + Send | ||
| where | ||
| R: AsyncFileReader + 'static; | ||
| } | ||
|
|
||
| #[derive(Clone, Default)] | ||
| pub struct NoopCache<K> { | ||
| #[derive(Default)] | ||
| pub struct NoCache<K> { | ||
| _phantom: PhantomData<K>, | ||
| } | ||
|
|
||
| impl<K> LruCache<K> for NoopCache<K> | ||
| where | ||
| K: Send + Sync + Clone + 'static, | ||
| { | ||
| type LruReader<R: AsyncFileReader + 'static> = R; | ||
|
|
||
| async fn new(_options: Options) -> Result<Self, Error> { | ||
| Ok(Self { | ||
| impl<K> Clone for NoCache<K> { | ||
| fn clone(&self) -> Self { | ||
| Self { | ||
| _phantom: PhantomData, | ||
| }) | ||
| } | ||
|
|
||
| async fn get_reader<R: AsyncFileReader>(&self, _key: K, reader: R) -> R { | ||
| reader | ||
| } | ||
| } | ||
| } | ||
|
|
||
| #[derive(Debug, Error)] | ||
| pub enum Error { | ||
| #[error("External lru implementation error: {0}")] | ||
| External(#[from] Box<dyn std::error::Error + Send + Sync + 'static>), | ||
| unsafe impl<K> Send for NoCache<K> {} | ||
|
|
||
| unsafe impl<K> Sync for NoCache<K> {} | ||
|
|
||
| impl<K> LruCache<K> for NoCache<K> | ||
| where | ||
| K: 'static, | ||
| { | ||
| type LruReader<R> = R | ||
| where | ||
| R: AsyncFileReader + 'static; | ||
|
|
||
| #[allow(clippy::manual_async_fn)] | ||
| fn get_reader<R>(&self, _key: K, reader: R) -> impl Future<Output = R> + Send | ||
| where | ||
| R: AsyncFileReader, | ||
| { | ||
| async move { reader } | ||
| } | ||
| } |
Oops, something went wrong.