Make TableProvider.scan() and PhysicalPlanner::create_physical_plan() async #1013

rdettai · 2021-09-17T08:59:46Z

Which issue does this PR close?

Closes #1012.

Rationale for this change

Simplify usage of async during the planning.
Support using DataFusion in systems where creating the scan requires network access (to, for example, find a list of files on some object store that match, or making a remote RPC call)

What changes are included in this PR?

mostly mechanical changes adding #[async_trait], async fn and .await annotations
Use BoxFuture for recursive async functions
Shorten the duration of the lock on the ExecutionContextState to avoid making it Send and propagate async to far in the API

Are there any user-facing changes?

API changes are relatively limited:

implementers of TableProvider need to make their implementation of scan async
ExecutionContext.create_physical_plan(&LogicalPlan) is now async and thus needs to be .awaited
~~BallistaContext.sql(&str) is now async and thus needs to be .awaited~~ -> Avoid creating ExecutionPlan for registered tables in Ballista #1016 fixed that

datafusion/src/execution/context.rs

rdettai · 2021-09-17T10:19:52Z

datafusion/src/physical_plan/planner.rs

+    fn create_initial_plan<'a>(
+        &'a self,
+        logical_plan: &'a LogicalPlan,
+        ctx_state: &'a ExecutionContextState,
+    ) -> BoxFuture<'a, Result<Arc<dyn ExecutionPlan>>> {


This shows as a big change but only because the function body needs to be wrapped into a BoxFuture with async { .. }.boxed()

why can't this function signature be made async fn create_initial_plan.... ? Why does it need a boxed future?

FWIW I found it easier to understand the diffs without whitespace: https://github.com/apache/arrow-datafusion/pull/1013/files?w=1

Because of async's internals: https://rust-lang.github.io/async-book/07_workarounds/04_recursion.html

Ah, I see -- so that it can recursively call itself. 👍

datafusion/tests/sql.rs

alamb · 2021-09-17T10:59:45Z

I would recommend adding another rationale for this change which would be to "Support using DataFusion in systems where creating the scan requires network access (to, for example, find a list of files on some object store that match, or making a remote RPC call)"

alamb

Thank you @rdettai -- I only skimmed the changes to ballista -- I looked carefully at the changes to DataFusion. I really like the change to make TableProvider.scan() async as it opens up a host of ways to plan that can use network / other async resources

I think we should remove the changes in parallelism for Extension nodes and Union nodes from this PR as they appear unrelated and I am less sure they are good change

I really like this change and I think it brings a lot of flexibility to TableProviders

However, given it is a non trivial API change, I think we should get buy in from other parties as well

cc @yjshen @houqp @Dandandan @andygrove

datafusion/src/execution/context.rs

alamb · 2021-09-17T11:09:26Z

datafusion/src/physical_plan/planner.rs

+    fn create_initial_plan<'a>(
+        &'a self,
+        logical_plan: &'a LogicalPlan,
+        ctx_state: &'a ExecutionContextState,
+    ) -> BoxFuture<'a, Result<Arc<dyn ExecutionPlan>>> {


why can't this function signature be made async fn create_initial_plan.... ? Why does it need a boxed future?

FWIW I found it easier to understand the diffs without whitespace: https://github.com/apache/arrow-datafusion/pull/1013/files?w=1

datafusion/src/physical_plan/planner.rs

rdettai · 2021-09-17T12:45:01Z

ballista/rust/client/src/context.rs

+                let execution_plan = ctx.create_physical_plan(&plan).await?;
                ctx.register_table(
                    TableReference::Bare { table: name },
                    Arc::new(DfTableAdapter::new(plan, execution_plan)),


I feel BallistaContext.sql(&str) should not become async. Opening #1016 to track that.

Dandandan · 2021-09-17T16:08:20Z

This looks like a useful change to me! I think we can benefit from the Tokio runtime for doing remote calls.
Also a lot of the planning could be done in parallel without too much changes - which is amazing too.

alamb · 2021-09-18T10:42:57Z

datafusion/src/physical_plan/planner.rs

+    fn create_initial_plan<'a>(
+        &'a self,
+        logical_plan: &'a LogicalPlan,
+        ctx_state: &'a ExecutionContextState,
+    ) -> BoxFuture<'a, Result<Arc<dyn ExecutionPlan>>> {


Ah, I see -- so that it can recursively call itself. 👍

yjshen · 2021-09-18T11:23:19Z

It is great to bring async to the planning phase. The changes are neat. I've tried to prototype this kind of staff before and find this easy to follow. Great job @rdettai 👍

rdettai · 2021-09-20T13:47:18Z

If we close #1028 first, I can rebase on that and avoid making BallistaContext.sql(&str) async

houqp · 2021-09-20T17:03:18Z

@rdettai you can rebase now :)

houqp · 2021-09-20T17:11:24Z

datafusion/src/execution/context.rs

+            // be written to. As for eventual modifications that would be applied to the
+            // original state after it has been cloned, they will not be picked up by the
+            // clone but that is okay, as it is equivalent to postponing the state update
+            // by keeping the lock until the end of the function scope.


👍 thanks for the clear write up

houqp

Thanks @rdettai for this great improvement. The change look good to me, good to go after we resolve the conflicts.

alamb · 2021-09-20T18:03:52Z

Any concerns with this change @Dandandan / @houqp ? If no one has any concerns, I would like to merge this as soon as it gets rebased and CI is passing

Thanks again @rdettai

houqp · 2021-09-20T18:05:06Z

No concern for me, +1 on merging it after CI is passing.

Dandandan · 2021-09-20T18:23:19Z

Also +1 from me

rdettai · 2021-09-21T07:07:32Z

Rebase on #1016 allowed to remove async from BallistaContext.sql(&str), only api changes are now:

implementers of TableProvider need to make their implementation of scan async
ExecutionContext.create_physical_plan(&LogicalPlan) is now async and thus needs to be .awaited

alamb · 2021-09-21T09:58:11Z

Thanks again @rdettai -- I think this is a great step forward towards supporting more object store / distributed planning systems with DataFusion ❤️ 🚀

rdettai · 2021-09-21T11:50:07Z

Thank you all for your review time! The next step of the journey is #1010. It not the support of the object store yet, but that will also come soon ! 😉

github-actions bot added ballista datafusion Changes in the datafusion crate labels Sep 17, 2021

rdettai marked this pull request as ready for review September 17, 2021 10:12

rdettai commented Sep 17, 2021

View reviewed changes

datafusion/src/execution/context.rs Show resolved Hide resolved

rdettai commented Sep 17, 2021

View reviewed changes

datafusion/tests/sql.rs Show resolved Hide resolved

alamb added the api change Changes the API exposed to users of the crate label Sep 17, 2021

alamb changed the title ~~Make TableProvider.scan() async~~ Make TableProvider.scan() and PhysicalPlanner::create_physical_plan() async Sep 17, 2021

alamb reviewed Sep 17, 2021

View reviewed changes

rdettai commented Sep 17, 2021

View reviewed changes

alamb approved these changes Sep 18, 2021

View reviewed changes

houqp reviewed Sep 20, 2021

View reviewed changes

rdettai added 4 commits September 21, 2021 08:41

[feat] make TableProvider.scan() async

4c012bc

[fix] python bindings

5d21b8b

[fix] phantom files

f764e0a

[fix] removed parallelization

df5be43

rdettai force-pushed the async-scan branch from ccb372c to df5be43 Compare September 21, 2021 06:57

[fix] remove async from sql after rebase

ebaa2c5

alamb merged commit 299ab7d into apache:master Sep 21, 2021

rdettai deleted the async-scan branch September 21, 2021 11:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Make TableProvider.scan() and PhysicalPlanner::create_physical_plan() async #1013

Make TableProvider.scan() and PhysicalPlanner::create_physical_plan() async #1013

rdettai commented Sep 17, 2021 •

edited

Loading

rdettai Sep 17, 2021

alamb Sep 17, 2021

rdettai Sep 17, 2021

alamb Sep 18, 2021

alamb commented Sep 17, 2021

alamb left a comment

alamb Sep 17, 2021

rdettai Sep 17, 2021 •

edited

Loading

Dandandan commented Sep 17, 2021

alamb Sep 18, 2021

yjshen commented Sep 18, 2021

rdettai commented Sep 20, 2021

houqp commented Sep 20, 2021

houqp Sep 20, 2021

houqp left a comment

alamb commented Sep 20, 2021

houqp commented Sep 20, 2021

Dandandan commented Sep 20, 2021

rdettai commented Sep 21, 2021

alamb commented Sep 21, 2021

rdettai commented Sep 21, 2021

Make TableProvider.scan() and PhysicalPlanner::create_physical_plan() async #1013

Make TableProvider.scan() and PhysicalPlanner::create_physical_plan() async #1013

Conversation

rdettai commented Sep 17, 2021 • edited Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

rdettai Sep 17, 2021

Choose a reason for hiding this comment

alamb Sep 17, 2021

Choose a reason for hiding this comment

rdettai Sep 17, 2021

Choose a reason for hiding this comment

alamb Sep 18, 2021

Choose a reason for hiding this comment

alamb commented Sep 17, 2021

alamb left a comment

Choose a reason for hiding this comment

alamb Sep 17, 2021

Choose a reason for hiding this comment

rdettai Sep 17, 2021 • edited Loading

Choose a reason for hiding this comment

Dandandan commented Sep 17, 2021

alamb Sep 18, 2021

Choose a reason for hiding this comment

yjshen commented Sep 18, 2021

rdettai commented Sep 20, 2021

houqp commented Sep 20, 2021

houqp Sep 20, 2021

Choose a reason for hiding this comment

houqp left a comment

Choose a reason for hiding this comment

alamb commented Sep 20, 2021

houqp commented Sep 20, 2021

Dandandan commented Sep 20, 2021

rdettai commented Sep 21, 2021

alamb commented Sep 21, 2021

rdettai commented Sep 21, 2021

rdettai commented Sep 17, 2021 •

edited

Loading

rdettai Sep 17, 2021 •

edited

Loading