[BugFix] add Metadata for PathExpand (#2720)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

Add Metadata for PathExpand, which is consistent with the Metadata of
`EdgeExpand` in `ExpandBase`.

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

Fixes #2714

interactive_engine/executor/ir/clients/rust/client/src/physical_builder.rs

@@ -451,6 +451,15 @@ impl JobBuilder {
         self
     }
 
+    // Notice that this is used to set the meta_data of the **Last Appended OP**
+    pub fn with_meta_data(&mut self, meta_data: Option<algebra_pb::MetaData>) {
+        self.plan.with_meta_data(
+            meta_data
+                .map(|meta| vec![meta.into()])
+                .unwrap_or_default(),
+        );
+    }
+
     pub fn sink(&mut self, sink: algebra_pb::Sink) {
         self.plan.sink(sink.into());
     }

interactive_engine/executor/ir/common/build.rs

@@ -30,6 +30,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
     println!("cargo:rerun-if-changed=../proto/schema.proto");
     println!("cargo:rerun-if-changed=../proto/results.proto");
     println!("cargo:rerun-if-changed=../proto/physical.proto");
+    println!("cargo:rerun-if-changed=../proto/type.proto");
     let out_dir = PathBuf::from(GEN_DIR);
     if out_dir.exists() {
         let _ = std::fs::remove_dir_all(GEN_DIR);
@@ -46,6 +47,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
                 "../proto/schema.proto",
                 "../proto/results.proto",
                 "../proto/physical.proto",
+                "../proto/type.proto",
             ],
             &["../proto"],
         )?;
@@ -61,6 +63,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
     println!("cargo:rerun-if-changed=../proto/schema.proto");
     println!("cargo:rerun-if-changed=../proto/results.proto");
     println!("cargo:rerun-if-changed=../proto/physical.proto");
+    println!("cargo:rerun-if-changed=../proto/type.proto");
     prost_build::Config::new()
         .type_attribute(".", "#[derive(Serialize,Deserialize)]")
         .compile_protos(
@@ -71,6 +74,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
                 "../proto/schema.proto",
                 "../proto/results.proto",
                 "../proto/physical.proto",
+                "../proto/type.proto",
             ],
             &["../proto"],
         )?;

interactive_engine/executor/ir/core/src/plan/physical.rs

@@ -247,6 +247,10 @@ impl AsPhysical for pb::PathExpand {
         if let Some(expand_base) = path_expand.base.as_mut() {
             let edge_expand = expand_base.edge_expand.as_mut();
             let getv = expand_base.get_v.as_mut();
+            let edge_expand_meta = edge_expand
+                .as_ref()
+                .map(|opr| opr.meta_data.clone())
+                .unwrap_or(None);
             if edge_expand.is_some() && getv.is_none() {
                 // Must be the case of EdgeExpand with Opt=Vertex
                 if edge_expand.unwrap().expand_opt != pb::edge_expand::ExpandOpt::Vertex as i32 {
@@ -286,10 +290,17 @@ impl AsPhysical for pb::PathExpand {
                     edge_expand, getv
                 )));
             }
+
+            path_expand.post_process(builder, plan_meta)?;
+            // Set the Metadata of PathExpand to the Metadata of EdgeExpand
+            builder
+                .path_expand(path_expand)
+                .with_meta_data(edge_expand_meta);
+
+            Ok(())
+        } else {
+            Err(IrError::MissingData("PathExpand::base".to_string()))
         }
-        path_expand.post_process(builder, plan_meta)?;
-        builder.path_expand(path_expand);
-        Ok(())
     }
 
     fn post_process(&mut self, builder: &mut JobBuilder, plan_meta: &mut PlanMeta) -> IrResult<()> {