Move physical plan serde from Ballista to DataFusion

datafusion/proto/Cargo.toml

@@ -41,9 +41,12 @@ json = ["pbjson", "serde", "serde_json"]
 
 [dependencies]
 arrow = "28.0.0"
+chrono = { version = "0.4", default-features = false }
 datafusion = { path = "../core", version = "14.0.0" }
 datafusion-common = { path = "../common", version = "14.0.0" }
 datafusion-expr = { path = "../expr", version = "14.0.0" }
+object_store = { version = "0.5.0" }
+parking_lot = { version = "0.12" }
 pbjson = { version = "0.5", optional = true }
 pbjson-types = { version = "0.5", optional = true }
 prost = "0.11.0"

datafusion/proto/README.md

@@ -46,9 +46,9 @@ fn main() -> Result<()> {
 }
 ```
 
-## Serializing Plans
+## Serializing Logical Plans
 
-Based on [examples/plan_serde.rs](examples/plan_serde.rs)
+Based on [examples/logical_plan_serde.rs](examples/logical_plan_serde.rs)
 
 ```rust
 use datafusion::prelude::*;
@@ -69,4 +69,29 @@ async fn main() -> Result<()> {
 }
 ```
 
+## Serializing Physical Plans
+
+Based on [examples/physical_plan_serde.rs](examples/physical_plan_serde.rs)
+
+```rust
+use datafusion::prelude::*;
+use datafusion_common::Result;
+use datafusion_proto::bytes::{physical_plan_from_bytes,physical_plan_to_bytes};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let ctx = SessionContext::new();
+    ctx.register_csv("t1", "testdata/test.csv", CsvReadOptions::default())
+        .await
+        ?;
+    let logical_plan = ctx.table("t1")?.to_logical_plan()?;
+    let physical_plan = ctx.create_physical_plan(&logical_plan).await?;
+    let bytes = physical_plan_to_bytes(physical_plan.clone())?;
+    let physical_round_trip = physical_plan_from_bytes(&bytes, &ctx)?;
+    assert_eq!(format!("{:?}", physical_plan), format!("{:?}", physical_round_trip));
+    Ok(())
+}
+
+```
+
 [df]: https://crates.io/crates/datafusion

datafusion/proto/examples/physical_plan_serde.rs

@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::prelude::*;
+use datafusion_common::Result;
+use datafusion_proto::bytes::{physical_plan_from_bytes, physical_plan_to_bytes};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let ctx = SessionContext::new();
+    ctx.register_csv("t1", "testdata/test.csv", CsvReadOptions::default())
+        .await?;
+    let logical_plan = ctx.table("t1")?.to_logical_plan()?;
+    let physical_plan = ctx.create_physical_plan(&logical_plan).await?;
+    let bytes = physical_plan_to_bytes(physical_plan.clone())?;
+    let physical_round_trip = physical_plan_from_bytes(&bytes, &ctx)?;
+    assert_eq!(
+        format!("{:?}", physical_plan),
+        format!("{:?}", physical_round_trip)
+    );
+    Ok(())
+}