Repository: alibaba/GraphScope (language: python)

Issue:
[BUG] GIE fails in filter edge by properties in some cases
**Describe the bug**

GIE fails in filter edge by specifying the condition in an expression, e.g.,
```
g.V().outE().where(expr("@.weight>0.5")).valueMap()
```
outputs unexpected results, that no edges is pruned.



Discussion / hints:

Fix patch:
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs b/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs
index e8cbf40f03b3..f9232d529602 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs
@@ -17,6 +17,7 @@ use std::convert::TryInto;
 
 use graph_proxy::apis::GraphElement;
 use graph_proxy::apis::{get_graph, DynDetails, GraphPath, QueryParams, Vertex};
+use graph_proxy::utils::expr::eval_pred::EvalPred;
 use ir_common::error::ParsePbError;
 use ir_common::generated::physical as pb;
 use ir_common::generated::physical::get_v::VOpt;
@@ -197,23 +198,27 @@ impl FilterMapFunction<Record, Record> for AuxiliaOperator {
                     }
                 }
                 EntryType::Edge => {
-                    // TODO: This is a little bit tricky. Modify this logic to query store with eid when supported.
-                    // Currently, when getting properties from an edge,
-                    // we assume that it has already been carried in the edge (when the first time queried the edge)
-                    // since on most storages, query edges by eid is not supported yet.
-                    if self.tag.eq(&self.alias) {
-                        // do nothing as we assume properties is already carried
-                    } else {
-                        let entry = entry.clone();
-                        if let Some(alias) = self.alias {
-                            // append without moving head
-                            input
-                                .get_columns_mut()
-                                .insert(alias as usize, entry);
-                        } else {
-                            input.append_arc_entry(entry, self.alias.clone());
+                    // TODO: This is a little bit tricky. Modify this logic to query store once query by eid is supported.
+                    // Currently, we support two cases:
+                    // 1. use auxilia to rename the edge to a new alias;
+                    // 2. use auxilia to filter the edge by predicates, where the necessary properties of the edge is assumed to be already pre-cached.
+                    let entry = entry.clone();
+                    if let Some(predicate) = &self.query_params.filter {
+                        let res = predicate
+                            .eval_bool(Some(&input))
+                            .map_err(|e| FnExecError::from(e))?;
+                        if !res {
+                            return Ok(None);
                         }
                     }
+                    if let Some(alias) = self.alias {
+                        // append without moving head
+                        input
+                            .get_columns_mut()
+                            .insert(alias as usize, entry);
+                    } else {
+                        input.append_arc_entry(entry, self.alias.clone());
+                    }
                 }
                 EntryType::Path => {
                     // Auxilia for vertices in Path is for filtering.
@@ -263,7 +268,6 @@ impl FilterMapFuncGen for pb::GetV {
                             .collect::<Result<Vec<_>, _>>()?;
                     }
                 }
-
                 let get_vertex_operator = GetVertexOperator {
                     start_tag: self.tag,
                     opt,


Test patch:
diff --git a/interactive_engine/executor/ir/integrated/tests/auxilia_test.rs b/interactive_engine/executor/ir/integrated/tests/auxilia_test.rs
index 78a02d071ae5..81bc7616b925 100644
--- a/interactive_engine/executor/ir/integrated/tests/auxilia_test.rs
+++ b/interactive_engine/executor/ir/integrated/tests/auxilia_test.rs
@@ -509,4 +509,48 @@ mod test {
         }
         assert_eq!(result_count, 6)
     }
+
+    // g.V().outE().where(expr("weight>0.5"))
+    #[test]
+    fn auxilia_edge_filter_test() {
+        let expand_opr = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: Some(query_params(vec![], vec!["weight".into()], None)),
+            expand_opt: 1, // edge
+            alias: None,
+        };
+
+        let auxilia_opr = pb::GetV {
+            tag: None,
+            opt: 4,
+            params: Some(query_params(vec![], vec![], str_to_expr_pb("@.weight>0.5".to_string()).ok())),
+            alias: Some(TAG_A.into()),
+        };
+
+        let conf = JobConf::new("auxilia_edge_filter_test");
+        let mut result = pegasus::run(conf, || {
+            let expand = expand_opr.clone();
+            let auxilia = auxilia_opr.clone();
+            |input, output| {
+                let mut stream = input.input_from(source_gen(None))?;
+                let flatmap_func = expand.gen_flat_map().unwrap();
+                stream = stream.flat_map(move |input| flatmap_func.exec(input))?;
+                let filter_map_func = auxilia.gen_filter_map().unwrap();
+                stream = stream.filter_map(move |input| filter_map_func.exec(input))?;
+                stream.sink_into(output)
+            }
+        })
+        .expect("build job failure");
+
+        let expected_ids = vec![2, 4];
+        let mut result_ids = vec![];
+        while let Some(Ok(record)) = result.next() {
+            if let Some(element) = record.get(Some(TAG_A)).unwrap().as_edge() {
+                result_ids.push(element.id());
+            }
+        }
+        result_ids.sort();
+        assert_eq!(result_ids, expected_ids)
+    }
 }