Repository: alibaba/GraphScope (language: python)

Issue:
Support Optional `EdgeExpand` in GIE Runtime
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]

As titled. Support an optional edge expand, which is used to further support optional match in queries. Basically, given a vertex $v$, optional edge expand is to expand its neighbors if any, or represents the absence of neighbors with `None`, while preserving such $v$ in the results.

This includes:

- [x] Add a field of `is_optional` in `EdgeExpand` in proto to express optional edge expand
- [x] Support a basic implementation of `OptionalEdgeExpand`
- [ ] Support more complex situations like `OptionalEdgeExpand` in `ExpandIntersection` and `PathExpand`

Besides, this will introduce `Null` value, we need to process it appropriately in other operations. E.g., expanding from a `Null` value may also leads to a `Null` result.

**Describe the solution you'd like**
A clear and concise description of what you want to happen.

**Describe alternatives you've considered**
A clear and concise description of any alternative solutions or features you've considered.

**Additional context**
Add any other context or screenshots about the feature request here.



Discussion / hints:

Fix patch:
diff --git a/docs/interactive_engine/neo4j/supported_cypher.md b/docs/interactive_engine/neo4j/supported_cypher.md
index 0a00bfbe9c9d..ac450fa1eecf 100644
--- a/docs/interactive_engine/neo4j/supported_cypher.md
+++ b/docs/interactive_engine/neo4j/supported_cypher.md
@@ -122,6 +122,14 @@ MATCH (a) -[]-> () -[]-> (b)  # second MATCH clause
 RETURN a, b;
 ```
 
+Besides, we support `OPTIONAL MATCH`. For example,
+the following query can be supported:
+```Cypher
+MATCH (a) -[]-> (b)
+Optional MATCH (b) -[]-> (c) 
+RETURN a, b, c;
+```
+
 | Keyword | Comments |  Supported  |  Todo
 |:---|---|:---:|:---|
 | MATCH | only one Match clause is allowed  | <input type="checkbox" disabled checked />  |
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java
index bfedfda21f49..cb00ef88a020 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java
@@ -209,4 +209,14 @@ public static QueryContext get_simple_match_query_16_test() {
         List<String> expected = Arrays.asList("Record<{$f0: 325593}>");
         return new QueryContext(query, expected);
     }
+
+    public static QueryContext get_simple_match_query_17_test() {
+        String query =
+                "MATCH (person:PERSON {id: 26388279067534})<-[:HASCREATOR]-(message: POST |"
+                        + " COMMENT)\n"
+                        + "OPTIONAL MATCH (message: POST | COMMENT)<-[like:LIKES]-(liker:PERSON)\n"
+                        + " Return count(person);";
+        List<String> expected = Arrays.asList("Record<{$f0: 851}>");
+        return new QueryContext(query, expected);
+    }
 }
diff --git a/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs b/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs
index 5a2a55eb2653..bb1c1a0d30f6 100644
--- a/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs
+++ b/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs
@@ -104,7 +104,7 @@ impl GraphPath {
                     let entry = entry.into();
                     let id = entry.id();
                     GraphPath::SimpleEndV((entry, vec![id], 1))
-                },
+                }
                 pb::path_expand::PathOpt::Trail => GraphPath::TrailAllPath(vec![entry.into()]),
             },
             pb::path_expand::ResultOpt::AllV | pb::path_expand::ResultOpt::AllVE => match path_opt {
@@ -167,21 +167,27 @@ impl GraphPath {
 
     pub fn get_path_start(&self) -> Option<&VertexOrEdge> {
         match self {
-            GraphPath::AllPath(ref p) | GraphPath::SimpleAllPath(ref p) | GraphPath::TrailAllPath(ref p) => p.first(),
+            GraphPath::AllPath(ref p)
+            | GraphPath::SimpleAllPath(ref p)
+            | GraphPath::TrailAllPath(ref p) => p.first(),
             GraphPath::EndV(_) | GraphPath::SimpleEndV(_) => None,
         }
     }
 
     pub fn get_path_end(&self) -> &VertexOrEdge {
         match self {
-            GraphPath::AllPath(ref p) | GraphPath::SimpleAllPath(ref p) | GraphPath::TrailAllPath(ref p) => p.last().unwrap(),
+            GraphPath::AllPath(ref p)
+            | GraphPath::SimpleAllPath(ref p)
+            | GraphPath::TrailAllPath(ref p) => p.last().unwrap(),
             GraphPath::EndV((ref e, _)) | GraphPath::SimpleEndV((ref e, _, _)) => e,
         }
     }
 
     pub fn get_path_end_mut(&mut self) -> &mut VertexOrEdge {
         match self {
-            GraphPath::AllPath(ref mut p) | GraphPath::SimpleAllPath(ref mut p) | GraphPath::TrailAllPath(ref mut p) => p.last_mut().unwrap(),
+            GraphPath::AllPath(ref mut p)
+            | GraphPath::SimpleAllPath(ref mut p)
+            | GraphPath::TrailAllPath(ref mut p) => p.last_mut().unwrap(),
             GraphPath::EndV((ref mut e, _)) | GraphPath::SimpleEndV((ref mut e, _, _)) => e,
         }
     }
@@ -233,7 +239,9 @@ impl GraphPath {
     // pop the last element from the path, and return the element.
     pub fn pop(&mut self) -> Option<VertexOrEdge> {
         match self {
-            GraphPath::AllPath(ref mut p) | GraphPath::SimpleAllPath(ref mut p) | GraphPath::TrailAllPath(ref mut p) => p.pop(),
+            GraphPath::AllPath(ref mut p)
+            | GraphPath::SimpleAllPath(ref mut p)
+            | GraphPath::TrailAllPath(ref mut p) => p.pop(),
             GraphPath::EndV(_) | GraphPath::SimpleEndV(_) => None,
         }
     }
@@ -241,7 +249,9 @@ impl GraphPath {
     // reverse the path.
     pub fn reverse(&mut self) {
         match self {
-            GraphPath::AllPath(ref mut p) | GraphPath::SimpleAllPath(ref mut p) | GraphPath::TrailAllPath(ref mut p) => {
+            GraphPath::AllPath(ref mut p)
+            | GraphPath::SimpleAllPath(ref mut p)
+            | GraphPath::TrailAllPath(ref mut p) => {
                 p.reverse();
             }
             GraphPath::EndV(_) | GraphPath::SimpleEndV(_) => {}
@@ -251,7 +261,9 @@ impl GraphPath {
     // get the element ids in the path, including both vertices and edges.
     pub fn get_elem_ids(&self) -> Vec<ID> {
         match self {
-            GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) | GraphPath::TrailAllPath(p) => p.iter().map(|e| e.id()).collect(),
+            GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) | GraphPath::TrailAllPath(p) => {
+                p.iter().map(|e| e.id()).collect()
+            }
             GraphPath::EndV((e, _)) | GraphPath::SimpleEndV((e, _, _)) => vec![e.id()],
         }
     }
@@ -259,7 +271,9 @@ impl GraphPath {
     // get the element labels in the path, including both vertices and edges.
     pub fn get_elem_labels(&self) -> Vec<Option<LabelId>> {
         match self {
-            GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) | GraphPath::TrailAllPath(p) => p.iter().map(|e| e.label()).collect(),
+            GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) | GraphPath::TrailAllPath(p) => {
+                p.iter().map(|e| e.label()).collect()
+            }
             GraphPath::EndV((e, _)) | GraphPath::SimpleEndV((e, _, _)) => vec![e.label()],
         }
     }
@@ -409,10 +423,10 @@ impl PartialEq for GraphPath {
             | (GraphPath::AllPath(p1), GraphPath::SimpleAllPath(p2))
             | (GraphPath::AllPath(p1), GraphPath::TrailAllPath(p2))
             | (GraphPath::SimpleAllPath(p1), GraphPath::AllPath(p2))
-            | (GraphPath::SimpleAllPath(p1), GraphPath::SimpleAllPath(p2)) 
-            | (GraphPath::SimpleAllPath(p1), GraphPath::TrailAllPath(p2)) 
-            | (GraphPath::TrailAllPath(p1), GraphPath::AllPath(p2)) 
-            | (GraphPath::TrailAllPath(p1), GraphPath::SimpleAllPath(p2)) 
+            | (GraphPath::SimpleAllPath(p1), GraphPath::SimpleAllPath(p2))
+            | (GraphPath::SimpleAllPath(p1), GraphPath::TrailAllPath(p2))
+            | (GraphPath::TrailAllPath(p1), GraphPath::AllPath(p2))
+            | (GraphPath::TrailAllPath(p1), GraphPath::SimpleAllPath(p2))
             | (GraphPath::TrailAllPath(p1), GraphPath::TrailAllPath(p2)) => p1.eq(p2),
             (GraphPath::EndV((p1, _)), GraphPath::EndV((p2, _)))
             | (GraphPath::EndV((p1, _)), GraphPath::SimpleEndV((p2, _, _)))
@@ -571,7 +585,9 @@ impl TryFrom<result_pb::GraphPath> for GraphPath {
 impl Hash for GraphPath {
     fn hash<H: Hasher>(&self, state: &mut H) {
         match self {
-            GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) | GraphPath::TrailAllPath(p) => p.hash(state),
+            GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) | GraphPath::TrailAllPath(p) => {
+                p.hash(state)
+            }
             GraphPath::EndV((e, _)) | GraphPath::SimpleEndV((e, _, _)) => e.hash(state),
         }
     }
diff --git a/interactive_engine/executor/ir/runtime/src/assembly.rs b/interactive_engine/executor/ir/runtime/src/assembly.rs
index 9cef1c501579..d6ea8040a923 100644
--- a/interactive_engine/executor/ir/runtime/src/assembly.rs
+++ b/interactive_engine/executor/ir/runtime/src/assembly.rs
@@ -786,7 +786,9 @@ impl<P: PartitionInfo, C: ClusterInfo> IRJobAssembly<P, C> {
                             base
                         )))
                     })?;
-                    if (pb::path_expand::ResultOpt::AllVE == unsafe { std::mem::transmute(path.result_opt) } || pb::path_expand::PathOpt::Trail == unsafe { std::mem::transmute(path.path_opt) })
+                    if (pb::path_expand::ResultOpt::AllVE
+                        == unsafe { std::mem::transmute(path.result_opt) }
+                        || pb::path_expand::PathOpt::Trail == unsafe { std::mem::transmute(path.path_opt) })
                         && pb::edge_expand::ExpandOpt::Vertex
                             == unsafe { std::mem::transmute(edge_expand.expand_opt) }
                     {
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/edge_expand.rs b/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/edge_expand.rs
index 8c5446f709e8..3b9104b38711 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/edge_expand.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/edge_expand.rs
@@ -15,6 +15,7 @@
 
 use std::convert::TryInto;
 
+use dyn_type::Object;
 use graph_proxy::apis::{
     get_graph, Direction, DynDetails, GraphElement, QueryParams, Statement, Vertex, ID,
 };
@@ -33,6 +34,7 @@ pub struct EdgeExpandOperator<E: Entry> {
     alias: Option<KeyId>,
     stmt: Box<dyn Statement<ID, E>>,
     expand_opt: ExpandOpt,
+    is_optional: bool,
 }
 
 impl<E: Entry + 'static> FlatMapFunction<Record, Record> for EdgeExpandOperator<E> {
@@ -43,32 +45,46 @@ impl<E: Entry + 'static> FlatMapFunction<Record, Record> for EdgeExpandOperator<
             match entry.get_type() {
                 EntryType::Vertex => {
                     let id = entry.id();
-                    let iter = self.stmt.exec(id)?;
+                    let mut iter = self.stmt.exec(id)?.peekable();
                     match self.expand_opt {
                         // the case of expand edge, and get end vertex;
                         ExpandOpt::Vertex => {
-                            let neighbors_iter = iter.map(|e| {
-                                if let Some(e) = e.as_edge() {
-                                    Vertex::new(
-                                        e.get_other_id(),
-                                        e.get_other_label().cloned(),
-                                        DynDetails::default(),
-                                    )
-                                } else {
-                                    unreachable!()
-                                }
-                            });
-                            Ok(Box::new(RecordExpandIter::new(
-                                input,
-                                self.alias.as_ref(),
-                                Box::new(neighbors_iter),
-                            )))
+                            if self.is_optional && iter.peek().is_none() {
+                                input.append(Object::None, self.alias);
+                                Ok(Box::new(vec![input].into_iter()))
+                            } else {
+                                let neighbors_iter = iter.map(|e| {
+                                    if let Some(e) = e.as_edge() {
+                                        Vertex::new(
+                                            e.get_other_id(),
+                                            e.get_other_label().cloned(),
+                                            DynDetails::default(),
+                                        )
+                                    } else {
+                                        unreachable!()
+                                    }
+                                });
+                                Ok(Box::new(RecordExpandIter::new(
+                                    input,
+                                    self.alias.as_ref(),
+                                    Box::new(neighbors_iter),
+                                )))
+                            }
                         }
                         // the case of expand neighbors, including edges/vertices
                         ExpandOpt::Edge => {
-                            Ok(Box::new(RecordExpandIter::new(input, self.alias.as_ref(), iter)))
+                            if self.is_optional && iter.peek().is_none() {
+                                input.append(Object::None, self.alias);
+                                Ok(Box::new(vec![input].into_iter()))
+                            } else {
+                                Ok(Box::new(RecordExpandIter::new(
+                                    input,
+                                    self.alias.as_ref(),
+                                    Box::new(iter),
+                                )))
+                            }
                         }
-                        // the case of get degree. TODO: this case should be a `Map`
+                        // the case of get degree.
                         ExpandOpt::Degree => {
                             let degree = iter.count();
                             input.append(object!(degree), self.alias);
@@ -77,12 +93,32 @@ impl<E: Entry + 'static> FlatMapFunction<Record, Record> for EdgeExpandOperator<
                     }
                 }
                 EntryType::Path => {
-                    let graph_path = entry
-                        .as_graph_path()
+                    if self.is_optional {
+                        Err(FnExecError::unsupported_error(
+                            "Have not supported Optional Edge Expand in Path entry yet",
+                        ))?
+                    } else {
+                        let graph_path = entry
+                            .as_graph_path()
+                            .ok_or_else(|| FnExecError::Unreachable)?;
+                        let iter = self.stmt.exec(graph_path.get_path_end().id())?;
+                        let curr_path = graph_path.clone();
+                        Ok(Box::new(RecordPathExpandIter::new(input, curr_path, iter)))
+                    }
+                }
+                EntryType::Object => {
+                    let obj = entry
+                        .as_object()
                         .ok_or_else(|| FnExecError::Unreachable)?;
-                    let iter = self.stmt.exec(graph_path.get_path_end().id())?;
-                    let curr_path = graph_path.clone();
-                    Ok(Box::new(RecordPathExpandIter::new(input, curr_path, iter)))
+                    if Object::None.eq(obj) {
+                        input.append(Object::None, self.alias);
+                        Ok(Box::new(vec![input].into_iter()))
+                    } else {
+                        Err(FnExecError::unexpected_data_error(&format!(
+                            "Cannot Expand from current entry {:?}",
+                            entry
+                        )))?
+                    }
                 }
                 _ => Err(FnExecError::unexpected_data_error(&format!(
                     "Cannot Expand from current entry {:?}",
@@ -99,9 +135,6 @@ impl FlatMapFuncGen for pb::EdgeExpand {
     fn gen_flat_map(
         self,
     ) -> FnGenResult<Box<dyn FlatMapFunction<Record, Record, Target = DynIter<Record>>>> {
-        if self.is_optional {
-            return Err(FnGenError::unsupported_error("optional edge expand in EdgeExpandOperator"));
-        }
         let graph = get_graph().ok_or_else(|| FnGenError::NullGraphError)?;
         let start_v_tag = self.v_tag;
         let edge_or_end_v_tag = self.alias;
@@ -127,6 +160,7 @@ impl FlatMapFuncGen for pb::EdgeExpand {
                         alias: edge_or_end_v_tag,
                         stmt,
                         expand_opt: ExpandOpt::Vertex,
+                        is_optional: self.is_optional,
                     };
                     Ok(Box::new(edge_expand_operator))
                 } else {
@@ -137,6 +171,7 @@ impl FlatMapFuncGen for pb::EdgeExpand {
                         alias: edge_or_end_v_tag,
                         stmt,
                         expand_opt: ExpandOpt::Edge,
+                        is_optional: self.is_optional,
                     };
                     Ok(Box::new(edge_expand_operator))
                 }
@@ -144,8 +179,13 @@ impl FlatMapFuncGen for pb::EdgeExpand {
             _ => {
                 // Expand edges or degree
                 let stmt = graph.prepare_explore_edge(direction, &query_params)?;
-                let edge_expand_operator =
-                    EdgeExpandOperator { start_v_tag, alias: edge_or_end_v_tag, stmt, expand_opt };
+                let edge_expand_operator = EdgeExpandOperator {
+                    start_v_tag,
+                    alias: edge_or_end_v_tag,
+                    stmt,
+                    expand_opt,
+                    is_optional: self.is_optional,
+                };
                 Ok(Box::new(edge_expand_operator))
             }
         }
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/unfold.rs b/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/unfold.rs
index 400dac9e39aa..b8bd8b5bb6f3 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/unfold.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/flatmap/unfold.rs
@@ -13,6 +13,7 @@
 //! See the License for the specific language governing permissions and
 //! limitations under the License.
 
+use dyn_type::Object;
 use graph_proxy::apis::{DynDetails, Element, Vertex};
 use ir_common::generated::physical as pb;
 use ir_common::KeyId;
@@ -57,29 +58,41 @@ impl FlatMapFunction<Record, Record> for UnfoldOperator {
                     .as_any_ref()
                     .downcast_ref::<IntersectionEntry>()
                 {
-                    let mut res = Vec::with_capacity(intersection.len());
-                    for item in intersection.iter().cloned() {
-                        let mut new_entry = input.clone();
-                        new_entry.append(Vertex::new(item, None, DynDetails::default()), self.alias);
-                        res.push(new_entry);
+                    let len = intersection.len();
+                    if len == 0 {
+                        input.append(Object::None, self.alias);
+                        Ok(Box::new(vec![input].into_iter()))
+                    } else {
+                        let mut res = Vec::with_capacity(len);
+                        for item in intersection.iter().cloned() {
+                            let mut new_entry = input.clone();
+                            new_entry.append(Vertex::new(item, None, DynDetails::default()), self.alias);
+                            res.push(new_entry);
+                        }
+                        Ok(Box::new(res.into_iter()))
                     }
-                    Ok(Box::new(res.into_iter()))
                 } else if let Some(general_intersection) = entry
                     .as_any_ref()
                     .downcast_ref::<GeneralIntersectionEntry>()
                 {
-                    let mut res = Vec::with_capacity(general_intersection.len());
-                    for (vid, matchings) in general_intersection.matchings_iter() {
-                        for matching in matchings {
-                            let mut new_entry = input.clone();
-                            for (column, tag) in matching {
-                                new_entry.append(column.clone(), Some(tag));
+                    let len = general_intersection.len();
+                    if len == 0 {
+                        input.append(Object::None, self.alias);
+                        Ok(Box::new(vec![input].into_iter()))
+                    } else {
+                        let mut res = Vec::with_capacity(len);
+                        for (vid, matchings) in general_intersection.matchings_iter() {
+                            for matching in matchings {
+                                let mut new_entry = input.clone();
+                                for (column, tag) in matching {
+                                    new_entry.append(column.clone(), Some(tag));
+                                }
+                                new_entry.append(Vertex::new(vid, None, DynDetails::default()), self.alias);
+                                res.push(new_entry);
                             }
-                            new_entry.append(Vertex::new(vid, None, DynDetails::default()), self.alias);
-                            res.push(new_entry);
                         }
+                        Ok(Box::new(res.into_iter()))
                     }
-                    Ok(Box::new(res.into_iter()))
                 } else {
                     Err(FnExecError::unexpected_data_error(
                         "downcast intersection entry in UnfoldOperator",
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 b07a23310bad..ef68632cab80 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
@@ -15,6 +15,7 @@
 
 use std::convert::TryInto;
 
+use dyn_type::Object;
 use graph_proxy::apis::GraphElement;
 use graph_proxy::apis::{get_graph, DynDetails, GraphPath, QueryParams, Vertex};
 use graph_proxy::utils::expr::eval_pred::EvalPred;
@@ -117,6 +118,16 @@ impl FilterMapFunction<Record, Record> for GetVertexOperator {
                 } else {
                     Err(FnExecError::unexpected_data_error("unreachable path end entry in GetV"))?
                 }
+            } else if let Some(obj) = entry.as_object() {
+                if Object::None.eq(obj) {
+                    input.append(Object::None, self.alias);
+                    Ok(Some(input))
+                } else {
+                    Err(FnExecError::unexpected_data_error(&format!(
+                        "Can only apply `GetV` on an object that is not None. The entry is {:?}",
+                        entry
+                    )))?
+                }
             } else {
                 Err(FnExecError::unexpected_data_error( &format!(
                     "Can only apply `GetV` (`Auxilia` instead) on an edge or path entry, while the entry is {:?}", entry
@@ -240,6 +251,28 @@ impl FilterMapFunction<Record, Record> for AuxiliaOperator {
                 } else {
                     return Ok(None);
                 }
+            } else if let Some(obj) = entry.as_object() {
+                if Object::None.eq(obj) {
+                    if let Some(predicate) = &self.query_params.filter {
+                        let res = predicate
+                            .eval_bool(Some(&input))
+                            .map_err(|e| FnExecError::from(e))?;
+                        if res {
+                            input.append(Object::None, self.alias);
+                            return Ok(Some(input));
+                        } else {
+                            return Ok(None);
+                        }
+                    } else {
+                        input.append(Object::None, self.alias);
+                        return Ok(Some(input));
+                    }
+                } else {
+                    Err(FnExecError::unexpected_data_error(&format!(
+                        "neither Vertex nor Edge entry is accessed in `Auxilia` operator, the entry is {:?}",
+                        entry
+                    )))?
+                }
             } else {
                 Err(FnExecError::unexpected_data_error(&format!(
                     "neither Vertex nor Edge entry is accessed in `Auxilia` operator, the entry is {:?}",


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java
index 49b42038d1b6..aa503ae020cc 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java
@@ -154,6 +154,14 @@ public void run_simple_match_16_test() {
         Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
     }
 
+    @Test
+    public void run_simple_match_17_test() {
+        assumeTrue("pegasus".equals(System.getenv("ENGINE_TYPE")));
+        QueryContext testQuery = SimpleMatchQueries.get_simple_match_query_17_test();
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
     @AfterClass
     public static void afterClass() {
         if (session != null) {
diff --git a/interactive_engine/executor/ir/integrated/tests/optional_expand_test.rs b/interactive_engine/executor/ir/integrated/tests/optional_expand_test.rs
new file mode 100644
index 000000000000..95abc4fee9d4
--- /dev/null
+++ b/interactive_engine/executor/ir/integrated/tests/optional_expand_test.rs
@@ -0,0 +1,342 @@
+//
+//! Copyright 2021 Alibaba Group Holding Limited.
+//!
+//! Licensed 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.
+//!
+//!
+
+mod common;
+
+#[cfg(test)]
+mod test {
+    use std::sync::Arc;
+
+    use dyn_type::Object;
+    use graph_proxy::apis::{register_graph, GraphElement};
+    use graph_proxy::create_exp_store;
+    use graph_store::ldbc::LDBCVertexParser;
+    use graph_store::prelude::DefaultId;
+    use ir_common::expr_parse::str_to_expr_pb;
+    use ir_common::generated::physical as pb;
+    use ir_common::KeyId;
+    use pegasus::api::{Map, Sink};
+    use pegasus::result::ResultStream;
+    use pegasus::JobConf;
+    use runtime::process::entry::Entry;
+    use runtime::process::operator::flatmap::FlatMapFuncGen;
+    use runtime::process::operator::map::FilterMapFuncGen;
+    use runtime::process::operator::source::SourceOperator;
+    use runtime::process::record::Record;
+
+    use crate::common::test::*;
+
+    // g.V()
+    fn source_gen(alias: Option<KeyId>) -> Box<dyn Iterator<Item = Record> + Send> {
+        source_gen_with_scan_opr(pb::Scan {
+            scan_opt: 0,
+            alias,
+            params: None,
+            idx_predicate: None,
+            is_count_only: false,
+        })
+    }
+
+    fn source_gen_with_scan_opr(scan_opr_pb: pb::Scan) -> Box<dyn Iterator<Item = Record> + Send> {
+        let graph = create_exp_store(Arc::new(TestCluster {}));
+        register_graph(graph);
+        let source = SourceOperator::new(scan_opr_pb.into(), Arc::new(TestRouter::default())).unwrap();
+        source.gen_source(0).unwrap()
+    }
+
+    fn expand_test(expand: pb::EdgeExpand) -> ResultStream<Record> {
+        let conf = JobConf::new("expand_test");
+        let result = pegasus::run(conf, || {
+            let expand = expand.clone();
+            |input, output| {
+                let mut stream = input.input_from(source_gen(Some(TAG_A)))?;
+                let flatmap_func = expand.gen_flat_map().unwrap();
+                stream = stream.flat_map(move |input| flatmap_func.exec(input))?;
+                stream.sink_into(output)
+            }
+        })
+        .expect("build job failure");
+        result
+    }
+
+    // g.V().out() with optional out
+    // v1, v4, v6 have out-neighbors; while v2, v3, v5 do not
+    #[test]
+    fn optional_expand_outv_test() {
+        let expand_opr_pb = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: None,
+            expand_opt: 0,
+            alias: None,
+            is_optional: true,
+        };
+        let mut result = expand_test(expand_opr_pb);
+        let mut result_ids = vec![];
+        let mut none_cnt = 0;
+        let v2: DefaultId = LDBCVertexParser::to_global_id(2, 0);
+        let v3: DefaultId = LDBCVertexParser::to_global_id(3, 1);
+        let v4: DefaultId = LDBCVertexParser::to_global_id(4, 0);
+        let v5: DefaultId = LDBCVertexParser::to_global_id(5, 1);
+        let mut expected_ids = vec![v2, v3, v3, v3, v4, v5];
+        while let Some(Ok(record)) = result.next() {
+            if let Some(element) = record.get(None).unwrap().as_vertex() {
+                result_ids.push(element.id() as usize)
+            } else if let Some(obj) = record.get(None).unwrap().as_object() {
+                assert_eq!(obj, &Object::None);
+                none_cnt += 1;
+            }
+        }
+        result_ids.sort();
+        expected_ids.sort();
+        assert_eq!(result_ids, expected_ids);
+        // v2, v3, v5 does not have out edges
+        assert_eq!(none_cnt, 3);
+    }
+
+    // g.V().out('knows') with optional out
+    // v1 has out knows neighbors of v2, v4; while other vertices do not have out knows neighbors
+    #[test]
+    fn optional_expand_outv_test_2() {
+        let query_param = query_params(vec![KNOWS_LABEL.into()], vec![], None);
+        let expand_opr_pb = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: Some(query_param),
+            expand_opt: 0,
+            alias: None,
+            is_optional: true,
+        };
+        let mut result = expand_test(expand_opr_pb);
+        let mut result_ids = vec![];
+        let mut none_cnt = 0;
+        let v2: DefaultId = LDBCVertexParser::to_global_id(2, 0);
+        let v4: DefaultId = LDBCVertexParser::to_global_id(4, 0);
+        let mut expected_ids = vec![v2, v4];
+        while let Some(Ok(record)) = result.next() {
+            println!("record: {:?}", record);
+            if let Some(element) = record.get(None).unwrap().as_vertex() {
+                result_ids.push(element.id() as usize)
+            } else if let Some(obj) = record.get(None).unwrap().as_object() {
+                assert_eq!(obj, &Object::None);
+                none_cnt += 1;
+            }
+        }
+        result_ids.sort();
+        expected_ids.sort();
+        assert_eq!(result_ids, expected_ids);
+        assert_eq!(none_cnt, 5);
+    }
+
+    // g.V().outE('knows', 'created') with optional out
+    #[test]
+    fn optional_expand_oute_with_many_labels_test() {
+        let query_param = query_params(vec![KNOWS_LABEL.into(), CREATED_LABEL.into()], vec![], None);
+        let expand_opr_pb = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: Some(query_param),
+            expand_opt: 1,
+            alias: None,
+            is_optional: true,
+        };
+        let mut result = expand_test(expand_opr_pb);
+        let mut result_edges = vec![];
+        let mut none_cnt = 0;
+        let v1: DefaultId = LDBCVertexParser::to_global_id(1, 0);
+        let v2: DefaultId = LDBCVertexParser::to_global_id(2, 0);
+        let v3: DefaultId = LDBCVertexParser::to_global_id(3, 1);
+        let v4: DefaultId = LDBCVertexParser::to_global_id(4, 0);
+        let v5: DefaultId = LDBCVertexParser::to_global_id(5, 1);
+        let v6: DefaultId = LDBCVertexParser::to_global_id(6, 0);
+        let mut expected_edges = vec![(v1, v2), (v1, v3), (v1, v4), (v4, v3), (v4, v5), (v6, v3)];
+        expected_edges.sort();
+        while let Some(Ok(record)) = result.next() {
+            if let Some(e) = record.get(None).unwrap().as_edge() {
+                result_edges.push((e.src_id as usize, e.dst_id as usize));
+            } else if let Some(obj) = record.get(None).unwrap().as_object() {
+                assert_eq!(obj, &Object::None);
+                none_cnt += 1;
+            }
+        }
+        result_edges.sort();
+        assert_eq!(result_edges, expected_edges);
+        assert_eq!(none_cnt, 3);
+    }
+
+    // g.V().out('knows').where(@ isnull) with optional out
+    // in this case, for the vertices v2, v3, v4, v5, v6, that do not have out knows edges.
+    #[test]
+    fn optional_expand_outv_filter_test() {
+        let query_param = query_params(vec![KNOWS_LABEL.into()], vec![], None);
+        let expand_opr_pb = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: Some(query_param),
+            expand_opt: 0,
+            alias: None,
+            is_optional: true,
+        };
+        let vertex_query_param = query_params(vec![], vec![], str_to_expr_pb("isnull @".to_string()).ok());
+        let auxilia_opr_pb = pb::GetV { tag: None, opt: 4, params: Some(vertex_query_param), alias: None };
+
+        let conf = JobConf::new("optional_expand_outv_filter_test");
+        let mut result = pegasus::run(conf, || {
+            let expand = expand_opr_pb.clone();
+            let auxilia = auxilia_opr_pb.clone();
+            |input, output| {
+                let mut stream = input.input_from(source_gen(Some(TAG_A)))?;
+                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 mut result_ids: Vec<usize> = vec![];
+        let v2: DefaultId = LDBCVertexParser::to_global_id(2, 0);
+        let v3: DefaultId = LDBCVertexParser::to_global_id(3, 1);
+        let v4: DefaultId = LDBCVertexParser::to_global_id(4, 0);
+        let v5: DefaultId = LDBCVertexParser::to_global_id(5, 1);
+        let v6: DefaultId = LDBCVertexParser::to_global_id(6, 0);
+        let expected_ids = vec![v2, v3, v4, v5, v6];
+        while let Some(Ok(record)) = result.next() {
+            let vertex = record
+                .get(Some(TAG_A))
+                .unwrap()
+                .as_vertex()
+                .unwrap();
+            result_ids.push(vertex.id() as usize);
+        }
+        assert_eq!(result_ids, expected_ids)
+    }
+
+    // g.V().outE('knows').inV() with optional outE
+    // in this case, for the vertices, e.g., v2, v3, v4, v5, v6, that do not have out knows edges, the result of inV() would also be regarded as None.
+    #[test]
+    fn optional_expand_oute_inv_test() {
+        let expand_opr = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: Some(query_params(vec![KNOWS_LABEL.into()], vec![], None)),
+            expand_opt: 1,
+            alias: None,
+            is_optional: true,
+        };
+
+        let getv_opr = pb::GetV {
+            tag: None,
+            opt: 1, // EndV
+            params: Some(query_params(vec![], vec![], None)),
+            alias: None,
+        };
+
+        let conf = JobConf::new("expand_oute_inv_test");
+        let mut result = pegasus::run(conf, || {
+            let expand = expand_opr.clone();
+            let getv = getv_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 = getv.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![];
+        let mut none_cnt = 0;
+        while let Some(Ok(record)) = result.next() {
+            if let Some(element) = record.get(None).unwrap().as_vertex() {
+                result_ids.push(element.id() as usize);
+            } else if let Some(obj) = record.get(None).unwrap().as_object() {
+                assert_eq!(obj, &Object::None);
+                none_cnt += 1;
+            } else {
+                unreachable!()
+            }
+        }
+        result_ids.sort();
+        assert_eq!(result_ids, expected_ids);
+        assert_eq!(none_cnt, 5);
+    }
+
+    // g.V().as(0).select(0).by(out().count().as(1)) with optional out
+    // in this case, the vertices that do not have out edges would also be taken into account in the count()
+    #[test]
+    fn optional_expand_out_degree_test() {
+        let conf = JobConf::new("expand_degree_fused_test");
+        let expand_opr_pb = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: None,
+            expand_opt: 2,
+            alias: Some(1.into()),
+            is_optional: true,
+        };
+        let getv = pb::GetV { tag: None, opt: 4, params: None, alias: Some(TAG_A) };
+        let project = pb::Project {
+            mappings: vec![pb::project::ExprAlias {
+                expr: str_to_expr_pb("@0".to_string()).ok(),
+                alias: None,
+            }],
+            is_append: true,
+        };
+
+        let mut pegasus_result = pegasus::run(conf, || {
+            let getv = getv.clone();
+            let expand = expand_opr_pb.clone();
+            let project = project.clone();
+            |input, output| {
+                let mut stream = input.input_from(source_gen(None))?;
+                let filter_map_func = getv.gen_filter_map().unwrap();
+                stream = stream.filter_map(move |input| filter_map_func.exec(input))?;
+                let flat_map_func = expand.gen_flat_map().unwrap();
+                stream = stream.flat_map(move |input| flat_map_func.exec(input))?;
+                let filter_map_func = project.gen_filter_map().unwrap();
+                stream = stream.filter_map(move |input| filter_map_func.exec(input))?;
+                stream.sink_into(output)
+            }
+        })
+        .expect("build job failure");
+
+        let mut results = vec![];
+        let v1: DefaultId = LDBCVertexParser::to_global_id(1, 0);
+        let v2: DefaultId = LDBCVertexParser::to_global_id(2, 0);
+        let v3: DefaultId = LDBCVertexParser::to_global_id(3, 1);
+        let v4: DefaultId = LDBCVertexParser::to_global_id(4, 0);
+        let v5: DefaultId = LDBCVertexParser::to_global_id(5, 1);
+        let v6: DefaultId = LDBCVertexParser::to_global_id(6, 0);
+        let mut expected_results = vec![(v1, 3), (v2, 0), (v3, 0), (v4, 2), (v5, 0), (v6, 1)];
+        while let Some(Ok(record)) = pegasus_result.next() {
+            if let Some(v) = record.get(None).unwrap().as_vertex() {
+                if let Some(degree_obj) = record.get(Some(1)).unwrap().as_object() {
+                    results.push((v.id() as DefaultId, degree_obj.as_u64().unwrap()));
+                }
+            }
+        }
+        results.sort();
+        expected_results.sort();
+
+        assert_eq!(results, expected_results)
+    }
+}