feat: datafusion based kernel engine #3831
Conversation
| let handler: Arc<dyn JsonHandler> = match self.handle.runtime_flavor() { | ||
| RuntimeFlavor::MultiThread => Arc::new(DefaultJsonHandler::new( | ||
| store, | ||
| Arc::new(TokioMultiThreadExecutor::new(self.handle.clone())), | ||
| )), | ||
| RuntimeFlavor::CurrentThread => Arc::new(DefaultJsonHandler::new( | ||
| store, | ||
| Arc::new(TokioBackgroundExecutor::new()), | ||
| )), | ||
| _ => panic!("unsupported runtime flavor"), | ||
| }; |
There was a problem hiding this comment.
We replicate this logic inside functions to avoid the generic parameter spilling out into our implementations. This tends to bubble up quite quickly since TaskExecutor is not dyn.
Once we move to datafusion specific implementations, these will not be generic over the runtime since datafusion and we are tied to tokio.
| Ok(Box::new( | ||
| grouped_files | ||
| .into_iter() | ||
| .map(|(url, files)| { | ||
| self.get_or_create_pq(url)?.read_parquet_files( | ||
| &files.to_vec(), | ||
| physical_schema.clone(), | ||
| predicate.clone(), | ||
| ) | ||
| }) | ||
| // TODO: this should not do any blocking operations, since this should | ||
| // happen when the iterators are polled and we are just creating a vec of iterators. | ||
| // Is this correct? | ||
| .try_collect::<_, Vec<_>, _>()? | ||
| .into_iter() | ||
| .flatten(), |
There was a problem hiding this comment.
Would love some feedback on this comment if I made the right assumptions here. We can also handle this in the map, but this gets a bit messy...
| pub(crate) fn group_by_store<T: IntoIterator<Item = impl AsObjectStoreUrl>>( | ||
| files: T, | ||
| ) -> std::collections::HashMap<ObjectStoreUrl, Vec<T::Item>> { | ||
| files | ||
| .into_iter() | ||
| .map(|item| (item.as_object_store_url(), item)) | ||
| .into_group_map() | ||
| } |
There was a problem hiding this comment.
This is the core piece for routing things to individual sub-instances scoped to an individual store.
There is a caveat in there in that we could be changing the order of inputs, which, of course, hurts log replay. But will look closer into this. The good news is that logs reside in a single store; only data files are potentially distributed across stores.
roeap
force-pushed
the
feat/kernel-df-engine
branch
from
ba9df85 to
86e7655
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #3831 +/- ##
==========================================
- Coverage 73.99% 73.74% -0.25%
==========================================
Files 148 151 +3
Lines 38904 39117 +213
Branches 38904 39117 +213
==========================================
+ Hits 28788 28848 +60
- Misses 8850 8999 +149
- Partials 1266 1270 +4 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Signed-off-by: Robert Pack <[email protected]>
roeap
force-pushed
the
feat/kernel-df-engine
branch
from
86e7655 to
7eeb6e6
Compare
rtyler
changed the title
2 participants
Description
This is the first direct step toward kernelized scans. The core of this will be a new
TableProvider. This new engine directly integrates with the datafusion session (more specifically,TaskContext). I thought about making a more generic engine that we could also use for non-datafusion code-paths. However, we would like to exchange the "inner" implementations to also leverage datafusion abstractions, at which point these would likely diverge. Thus, hoping to integrate this deeper first, before seeing if and how we can share between implementations.Why do we need a dedicated engine? URL support and moving away from object store paths.
I have this integrated with a table provider locally, but I am trying to keep things reviewable.