Skip to content

Commit f2e2c29

Browse files
authored
Move physical plan serde from Ballista to DataFusion (#4390)
* Move physical plan serde from Ballista to DataFusion Signed-off-by: kikkon <[email protected]> * add license * rebase master * fix * cargo check again * tomlfmt * remove ballista * remove ballista * resolve comment * remove ballista Signed-off-by: kikkon <[email protected]>
1 parent 23b4495 commit f2e2c29

14 files changed

Lines changed: 15336 additions & 3253 deletions

File tree

‎datafusion/proto/Cargo.toml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,9 +41,12 @@ json = ["pbjson", "serde", "serde_json"]
4141

4242
[dependencies]
4343
arrow = "28.0.0"
44+
chrono = { version = "0.4", default-features = false }
4445
datafusion = { path = "../core", version = "14.0.0" }
4546
datafusion-common = { path = "../common", version = "14.0.0" }
4647
datafusion-expr = { path = "../expr", version = "14.0.0" }
48+
object_store = { version = "0.5.0" }
49+
parking_lot = { version = "0.12" }
4750
pbjson = { version = "0.5", optional = true }
4851
pbjson-types = { version = "0.5", optional = true }
4952
prost = "0.11.0"

‎datafusion/proto/README.md‎

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,9 @@ fn main() -> Result<()> {
4646
}
4747
```
4848

49-
## Serializing Plans
49+
## Serializing Logical Plans
5050

51-
Based on [examples/plan_serde.rs](examples/plan_serde.rs)
51+
Based on [examples/logical_plan_serde.rs](examples/logical_plan_serde.rs)
5252

5353
```rust
5454
use datafusion::prelude::*;
@@ -69,4 +69,29 @@ async fn main() -> Result<()> {
6969
}
7070
```
7171

72+
## Serializing Physical Plans
73+
74+
Based on [examples/physical_plan_serde.rs](examples/physical_plan_serde.rs)
75+
76+
```rust
77+
use datafusion::prelude::*;
78+
use datafusion_common::Result;
79+
use datafusion_proto::bytes::{physical_plan_from_bytes,physical_plan_to_bytes};
80+
81+
#[tokio::main]
82+
async fn main() -> Result<()> {
83+
let ctx = SessionContext::new();
84+
ctx.register_csv("t1", "testdata/test.csv", CsvReadOptions::default())
85+
.await
86+
?;
87+
let logical_plan = ctx.table("t1")?.to_logical_plan()?;
88+
let physical_plan = ctx.create_physical_plan(&logical_plan).await?;
89+
let bytes = physical_plan_to_bytes(physical_plan.clone())?;
90+
let physical_round_trip = physical_plan_from_bytes(&bytes, &ctx)?;
91+
assert_eq!(format!("{:?}", physical_plan), format!("{:?}", physical_round_trip));
92+
Ok(())
93+
}
94+
95+
```
96+
7297
[df]: https://crates.io/crates/datafusion
File renamed without changes.
Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use datafusion::prelude::*;
19+
use datafusion_common::Result;
20+
use datafusion_proto::bytes::{physical_plan_from_bytes, physical_plan_to_bytes};
21+
22+
#[tokio::main]
23+
async fn main() -> Result<()> {
24+
let ctx = SessionContext::new();
25+
ctx.register_csv("t1", "testdata/test.csv", CsvReadOptions::default())
26+
.await?;
27+
let logical_plan = ctx.table("t1")?.to_logical_plan()?;
28+
let physical_plan = ctx.create_physical_plan(&logical_plan).await?;
29+
let bytes = physical_plan_to_bytes(physical_plan.clone())?;
30+
let physical_round_trip = physical_plan_from_bytes(&bytes, &ctx)?;
31+
assert_eq!(
32+
format!("{:?}", physical_plan),
33+
format!("{:?}", physical_round_trip)
34+
);
35+
Ok(())
36+
}

0 commit comments

Comments
 (0)