Improvements to Ballista extensibility

[thinkharderdev]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
(This section helps Arrow developers understand the context and why for this feature, in addition to the what)

Currently, we are working with DataFusion/Ballista as a query execution engine. One of the primary selling points for DataFusion is extensibility but it is not currently possible to use the many extension points in DataFusion with Ballista.

This is primarily due to the constraints of serializing all logical and physical plans as Protobuf messages.

Ideally we would like to use Ballista to execute:

• Scans using custom object stores
• User Defined logical plan extensions
• User defined physical plan extensions
• User defined scalar and aggregation functions

Describe the solution you'd like
A clear and concise description of what you want to happen.

There are two things ideally:

1. We would like to decouple the core Ballista functionality from the serializable representations of plans so that the serde layer can become pluggable/extensible.
2. Serde should be aware of a user-defined ExecutionContext so we can leverage optimizers, extension planners, and udf/udaf

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

There currently is no workaround for this but we have been prototyping possible solutions which we'd be interested in upstreaming.

Additional context
Add any other context or screenshots about the feature request here.

[alamb]

cc @realno @gaojun2048 @yahoNanJing

[andygrove]

It may be useful to see how substrait is handling extensions as well - https://substrait.io/extensions/

[realno]

This would be a great improvement 👍 I will follow the design and PRs

[yahoNanJing]

Thanks @thinkharderdev for proposing these potentials.

Scans using custom object stores

For this, actually our team has implemented for the HDFS. To avoid new object store registration, our workaround is to make the path self description with its scheme, like hdfs:://localhost:15050/..../file.parquet. Then with the scheme, we will know which kind of remote object store we needs.

User Defined logical plan extensions, physical plan extensions, scalar and aggregation functions

Maybe it's better to introduce the substrait integration into the roadmap.

[realno]

Maybe it's better to introduce the substrait integration into the roadmap.

+1

[houqp]

FWIW, Andy wrote a substrait rust implementation: https://github.com/andygrove/substrait-rs.

[thinkharderdev]

Agree on the substrait integration. It would definitely be nice to have a universal serializable representation and a way to configure extensions delcaritively.

I posted a draft PR apache/datafusion#1677 which I think can solve the immediate term issues with extensibility and also will be useful in migrating to a substrait-based implementation. By decoupling the representation from the core execution engine we can avoid a "Big Bang" migration (not to mention and endless parade of painful rebases while in development :))

[alamb]

I had a comment on apache/datafusion#1677 (review) which I think is worth considering

[thinkharderdev]

Related questions after tinkering a bit more today:

Should SchemaProvider methods be async? It could be useful to support integration with external metadata catalogs (AWS Glue, etc). This could also simplify the serialization of LogicalPlan::TableScans by just passing a TableReference and resolving the TableProvider using the context.

[alamb]

Should SchemaProvider methods be async? It could be useful to support integration with external metadata catalogs (AWS Glue, etc). This could also simplify the serialization of LogicalPlan::TableScans by just passing a TableReference and resolving the TableProvider using the context.

I can see a rationale for making schema provider methods async if the usecase is a one time query across a pile of parquet files.

However, in general if one has to do network IO to figure out what tables exist or their schemas, it may be hard to get adequate performance

[EricJoy2048]

cc @realno @gaojun2048 @yahoNanJing

Thank's for your CC.
I learned substrait documentation, I think it's useful to make plan serializable pluggable/extensible.

[thinkharderdev]

I think all the points in this have been addressed so we can close this issue.