phillipleblanc commented on issue #2249:
URL:
https://github.com/apache/datafusion-ballista/issues/2249#issuecomment-5235014711
I think we could reframe this as extending Ballista with:
1. Pluggable, host-mountable query frontends.
2. Pluggable plan-input decoders and relation translators.
3. A transport-neutral query execution API for Ballista and other embedders.
Spark Connect could be the first new frontend. The existing Ballista gRPC
service would become the default native frontend.
For context, Ballista already accepts Substrait plans, but the handling is
hardcoded into `ExecuteQueryParams` and `grpc.rs`. We could add Spark Connect
as another protobuf arm, but that would further couple Ballista's native RPC
protocol to every input protocol and plan representation.
The important distinction is:
- Substrait is a serialized plan format: bytes -> DataFusion `LogicalPlan`.
- Spark Connect is a stateful RPC protocol covering plans, sessions,
configuration, commands, artifacts, UDFs, cancellation, and result handling.
Substrait is therefore a plan decoder used by a frontend, rather than a
frontend itself. Spark Connect requires both a protocol frontend and a
relation/expression translator.
Something like:
```text
Spark Connect / Ballista gRPC / Flight SQL / future protocols
|
mountable protocol frontend
|
plan decoder / relation translator
|
DataFusion LogicalPlan
|
transport-neutral QueryBackend
/ \
BallistaQueryBackend embedding-specific backend
distributed jobs e.g. Spice policy/cache/routing
```
The frontend should not depend directly on `SchedulerServer` internals. It
should target a transport-neutral execution contract covering the entire query
lifecycle:
```rust
#[async_trait]
pub trait QueryBackend: Send + Sync {
async fn open_session(
&self,
request: SessionRequest,
) -> Result<QuerySession>;
async fn execute_plan(
&self,
request: ExecutePlanRequest,
) -> Result<Arc<dyn QueryOperation>>;
}
#[async_trait]
pub trait QueryOperation: Send + Sync {
fn id(&self) -> &str;
async fn status(&self) -> Result<QueryStatus>;
async fn cancel(&self) -> Result<()>;
async fn result_stream(&self) -> Result<SendableRecordBatchStream>;
}
```
Ballista would provide a default `BallistaQueryBackend`. Embedders could
wrap either backend with their own authorization, validation, caching, tracing,
and execution-routing policies.
Frontends should also be mountable into a server owned by the embedding
application, rather than requiring Ballista to own the listener:
```rust
pub trait GrpcQueryFrontend: Send + Sync {
fn register(
&self,
routes: &mut tonic::service::RoutesBuilder,
backend: Arc<dyn QueryBackend>,
) -> Result<()>;
}
```
Spice currently embeds `SchedulerServer`, owns the Tonic servers and
middleware, constructs custom Ballista sessions/codecs/state, assigns UUIDv7
job IDs, and directly consumes Ballista status events and result partitions. It
also applies Spice-specific validation, caching, authorization, and tracing
before deciding whether to execute locally or through Ballista.
With this model, Spice could:
- Mount the native Ballista and Spark Connect services into its existing
server.
- Implement `SpiceQueryBackend` so Spark Connect follows Spice's existing
query policies.
- Replace direct access to `SchedulerServer.state` with stable
`SchedulerHandle` and `JobHandle` APIs.
- Continue supplying UUIDv7 job IDs and recovering jobs across schedulers
without fork-specific submission APIs.
The embedded Ballista API would therefore need to support something like:
```rust
SchedulerHandle::create_session(...)
SchedulerHandle::submit(SubmitRequest {
requested_job_id: Some(job_id),
...
})
SchedulerHandle::recover(job_id)
SchedulerHandle::job(job_id) -> JobHandle
```
`JobHandle` would expose status, event subscription, cancellation,
completion, and output retrieval.
Separately, within the native Ballista gRPC frontend, we could make plan
decoding extensible. For backward compatibility, retain the existing protobuf
fields and add a generic encoded-plan envelope:
```proto
message EncodedQuery {
string format = 1; // e.g. "substrait.plan.v1"
bytes payload = 2;
map<string, string> options = 3;
}
oneof query {
bytes logical_plan = 1;
bytes substrait_plan = 6;
bytes physical_plan = 7;
EncodedQuery encoded_query = 8;
}
```
The existing fields would route through built-in decoders:
```rust
#[async_trait]
pub trait QueryInputDecoder: Send + Sync {
fn format(&self) -> &'static str;
async fn decode(
&self,
payload: &[u8],
session: &SessionContext,
options: &HashMap<String, String>,
) -> Result<LogicalPlan>;
}
```
DataFusion protobuf and Substrait would become built-in implementations.
Additional plan formats could be registered without modifying `grpc.rs` or
allocating a new protobuf field each time.
If this seems useful, this is something I could help shepherd through - as
Spice could start to reduce the reliance on our custom fork features in favor
of this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
