Repository: alibaba/GraphScope (language: python)

Issue:
[BUG] A bug in pkscan with expected values given in `Within`
**Describe the bug**
A clear and concise description of what the bug is.

When querying `g.V().hasLabel("Person").has("id", P.within(1,2,3))` where "id" is a primary key, Runtime wrongly parsed it into a multi-column primary key value. However, "1,2,3" are single-column primary key values, with "OR" semantics.



Discussion / hints:

Fix patch:
diff --git a/interactive_engine/executor/ir/common/src/utils.rs b/interactive_engine/executor/ir/common/src/utils.rs
index 49076c65b09a..ea646339e706 100644
--- a/interactive_engine/executor/ir/common/src/utils.rs
+++ b/interactive_engine/executor/ir/common/src/utils.rs
@@ -437,6 +437,13 @@ impl TryFrom<pb::IndexPredicate> for Vec<Vec<(NameOrId, Object)>> {
     type Error = ParsePbError;
 
     fn try_from(value: pb::IndexPredicate) -> Result<Self, Self::Error> {
+        // transform the `IndexPredicate` to `Vec<Vec<(NameOrId, Object)>>`, e.g.,
+        // a IndexPredicate can be: name="marko" && age=29 || name="josh" && age=27, which is an OrCondition with two AndConditions,
+        // then the result will be: [[("name", "marko"), ("age", 29)], [("name", "josh"), ("age", 27)]].
+        // Specifically, when the `IndexPredicate` contains a `within` condition, e.g.,
+        // a IndexPredicate can be: name within ["marko", "josh"], which is a single AndCondition, but with "OR" semantics,
+        // then the result should be: [[("name", "marko")], [("name", "josh")]].
+        // But if the IndexPredicate mix up with "within" and other conditions, unexpected result will be returned.
         let mut primary_key_values = Vec::with_capacity(value.or_predicates.len());
         for and_predicates in value.or_predicates {
             // PkValue can be one-column or multi-columns, which is a set of and_conditions.
@@ -462,26 +469,27 @@ impl TryFrom<pb::IndexPredicate> for Vec<Vec<(NameOrId, Object)>> {
                 if let pb::index_predicate::triplet::Value::Const(value) = value_pb {
                     if let Some(item) = value.item.as_ref() {
                         if cmp.eq(&common_pb::Logical::Within) {
-                            // specifically, if the cmp is within, the value must be an array
+                            // specifically, if the cmp is within, the value must be an array,
+                            // and it is "OR" semantics
                             match item {
                                 common_pb::value::Item::I32Array(array) => {
                                     for v in array.item.iter() {
-                                        primary_key_value.push((key.clone(), (*v).into()));
+                                        primary_key_values.push(vec![(key.clone(), (*v).into())]);
                                     }
                                 }
                                 common_pb::value::Item::I64Array(array) => {
                                     for v in array.item.iter() {
-                                        primary_key_value.push((key.clone(), (*v).into()));
+                                        primary_key_values.push(vec![(key.clone(), (*v).into())]);
                                     }
                                 }
                                 common_pb::value::Item::F64Array(array) => {
                                     for v in array.item.iter() {
-                                        primary_key_value.push((key.clone(), (*v).into()));
+                                        primary_key_values.push(vec![(key.clone(), (*v).into())]);
                                     }
                                 }
                                 common_pb::value::Item::StrArray(array) => {
                                     for v in array.item.iter() {
-                                        primary_key_value.push((key.clone(), (v.clone()).into()));
+                                        primary_key_values.push(vec![(key.clone(), (v.clone()).into())]);
                                     }
                                 }
                                 _ => Err(ParsePbError::ParseError(format!(
@@ -502,7 +510,9 @@ impl TryFrom<pb::IndexPredicate> for Vec<Vec<(NameOrId, Object)>> {
                     )))?
                 }
             }
-            primary_key_values.push(primary_key_value);
+            if !primary_key_value.is_empty() {
+                primary_key_values.push(primary_key_value);
+            }
         }
         Ok(primary_key_values)
     }
diff --git a/interactive_engine/executor/ir/graph_proxy/src/adapters/exp_store/read_graph.rs b/interactive_engine/executor/ir/graph_proxy/src/adapters/exp_store/read_graph.rs
index 933b0d42a250..32b7362db4cf 100644
--- a/interactive_engine/executor/ir/graph_proxy/src/adapters/exp_store/read_graph.rs
+++ b/interactive_engine/executor/ir/graph_proxy/src/adapters/exp_store/read_graph.rs
@@ -26,7 +26,7 @@ use graph_store::prelude::{
     DefaultId, EdgeId, GlobalStoreTrait, GlobalStoreUpdate, GraphDBConfig, InternalId, LDBCGraphSchema,
     LargeGraphDB, LocalEdge, LocalVertex, MutableGraphDB, Row, INVALID_LABEL_ID,
 };
-use ir_common::{KeyId, LabelId, NameOrId};
+use ir_common::{KeyId, LabelId, NameOrId, OneOrMany};
 use pegasus::configure_with_default;
 use pegasus_common::downcast::*;
 use pegasus_common::impl_as_any;
@@ -257,11 +257,34 @@ impl ReadGraph for ExpStore {
     }
 
     fn index_scan_vertex(
-        &self, _label: LabelId, _primary_key: &PKV, _params: &QueryParams,
+        &self, label: LabelId, primary_key: &PKV, params: &QueryParams,
     ) -> GraphProxyResult<Option<Vertex>> {
-        Err(GraphProxyError::unsupported_error(
-            "Experiment storage does not support index_scan_vertex for now",
-        ))?
+        let worker_idx = self.cluster_info.get_worker_index()?;
+        let workers_num = self.cluster_info.get_local_worker_num()?;
+        if worker_idx % workers_num == 0 {
+            let store_indexed_id = match primary_key {
+                OneOrMany::One(pkv) => encode_store_pk_prop_val(pkv[0].1.clone())?,
+                OneOrMany::Many(pkvs) => {
+                    if pkvs.len() == 1 {
+                        encode_store_pk_prop_val(pkvs[0].1.clone())?
+                    } else {
+                        return Err(GraphProxyError::unsupported_error(&format!(
+                            "Experiment storage does not support multiple primary keys {:?}",
+                            pkvs
+                        )))?;
+                    }
+                }
+            };
+            let gid: DefaultId = LDBCVertexParser::to_global_id(store_indexed_id, label as StoreLabelId);
+            if let Some(local_vertex) = self.store.get_vertex(gid) {
+                let v = to_runtime_vertex(local_vertex, params.columns.clone());
+                Ok(Some(v))
+            } else {
+                Ok(None)
+            }
+        } else {
+            Ok(None)
+        }
     }
 
     fn scan_edge(&self, params: &QueryParams) -> GraphProxyResult<Box<dyn Iterator<Item = Edge> + Send>> {
@@ -610,6 +633,24 @@ fn encode_storage_label(labels: &Vec<LabelId>) -> Option<Vec<StoreLabelId>> {
     }
 }
 
+// Specifically, exp_store only support id as pk
+fn encode_store_pk_prop_val(val: Object) -> GraphProxyResult<DefaultId> {
+    match val {
+        Object::Primitive(v) => v.as_usize().map_err(|_err| {
+            GraphProxyError::unsupported_error(&format!(
+                "Experiment storage does not support non-integer primary key {:?}",
+                v
+            ))
+        }),
+        _ => {
+            return Err(GraphProxyError::unsupported_error(&format!(
+                "Experiment storage does not support non-primitive primary key {:?}",
+                val
+            )))
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use graph_store::common::LabelId;


Test patch:
diff --git a/interactive_engine/executor/ir/integrated/tests/scan_test.rs b/interactive_engine/executor/ir/integrated/tests/scan_test.rs
index 30f3e4463355..75fa6d525f25 100644
--- a/interactive_engine/executor/ir/integrated/tests/scan_test.rs
+++ b/interactive_engine/executor/ir/integrated/tests/scan_test.rs
@@ -351,4 +351,71 @@ mod test {
         }
         assert_eq!(result, expected)
     }
+
+    // g.V().hasLabel('person').has("id", 1)
+    #[test]
+    fn scan_pk_test() {
+        let source_iter = scan_gen(pb::Scan {
+            scan_opt: 0,
+            alias: None,
+            params: Some(query_params(vec![PERSON_LABEL.into()], vec![], None)),
+            idx_predicate: Some(ir_common::generated::algebra::IndexPredicate {
+                or_predicates: vec![ir_common::generated::algebra::index_predicate::AndPredicate {
+                    predicates: vec![ir_common::generated::algebra::index_predicate::Triplet {
+                        key: Some("id".to_string().into()),
+                        value: Some(1.into()),
+                        cmp: 0, // eq
+                    }],
+                }],
+            }),
+            is_count_only: false,
+        });
+        let mut result_ids = vec![];
+        let v1: DefaultId = LDBCVertexParser::to_global_id(1, 0);
+        let mut expected_ids = vec![v1];
+        for record in source_iter {
+            if let Some(element) = record.get(None).unwrap().as_vertex() {
+                result_ids.push(element.id() as usize)
+            }
+        }
+        result_ids.sort();
+        expected_ids.sort();
+        assert_eq!(result_ids, expected_ids)
+    }
+
+    // g.V().hasLabel('person').has("id", P.within(1,2,4,6))
+    #[test]
+    fn scan_pk_within_test() {
+        let source_iter = scan_gen(pb::Scan {
+            scan_opt: 0,
+            alias: None,
+            params: Some(query_params(vec![PERSON_LABEL.into()], vec![], None)),
+            idx_predicate: Some(ir_common::generated::algebra::IndexPredicate {
+                or_predicates: vec![ir_common::generated::algebra::index_predicate::AndPredicate {
+                    predicates: vec![ir_common::generated::algebra::index_predicate::Triplet {
+                        key: Some("id".to_string().into()),
+                        value: Some(ir_common::generated::algebra::index_predicate::triplet::Value::Const(
+                            vec![1, 2, 4, 6].into(),
+                        )),
+                        cmp: 6, // within
+                    }],
+                }],
+            }),
+            is_count_only: false,
+        });
+        let mut result_ids = vec![];
+        let v1: DefaultId = LDBCVertexParser::to_global_id(1, 0);
+        let v2: DefaultId = LDBCVertexParser::to_global_id(2, 0);
+        let v4: DefaultId = LDBCVertexParser::to_global_id(4, 0);
+        let v6: DefaultId = LDBCVertexParser::to_global_id(6, 0);
+        let mut expected_ids = vec![v1, v2, v4, v6];
+        for record in source_iter {
+            if let Some(element) = record.get(None).unwrap().as_vertex() {
+                result_ids.push(element.id() as usize)
+            }
+        }
+        result_ids.sort();
+        expected_ids.sort();
+        assert_eq!(result_ids, expected_ids)
+    }
 }