Repository: dbpunk-labs/db3 (language: rust)

Issue:
Support multi-key index and support first key eq and combine key eq filter
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->

- Would you like to implement this feature? [y/n]

## Solution

<!-- Describe the solution you'd like. -->

## Alternatives

<!-- Describe 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/src/base/src/bson_util.rs b/src/base/src/bson_util.rs
index 9211fe20..6e0818e7 100644
--- a/src/base/src/bson_util.rs
+++ b/src/base/src/bson_util.rs
@@ -1,16 +1,18 @@
-use bson::Bson;
 use bson::Document;
 use bson::RawDocumentBuf;
+use bson::{Array, Bson};
 use byteorder::{BigEndian, WriteBytesExt};
 use db3_error::DB3Error;
+use db3_proto::db3_database_proto::structured_query::composite_filter::Operator as CompositeOp;
 use db3_proto::db3_database_proto::structured_query::field_filter::Operator;
 use db3_proto::db3_database_proto::structured_query::filter::FilterType;
 use db3_proto::db3_database_proto::structured_query::value::ValueType;
-use db3_proto::db3_database_proto::structured_query::FieldFilter;
 use db3_proto::db3_database_proto::structured_query::Filter;
 use db3_proto::db3_database_proto::structured_query::Value;
+use db3_proto::db3_database_proto::structured_query::{CompositeFilter, FieldFilter};
 use db3_proto::db3_database_proto::{index::IndexField, Index};
 use serde_json::Value as JsonValue;
+
 /// convert json string to Bson::Document
 pub fn json_str_to_bson_document(json_str: &str) -> std::result::Result<Document, DB3Error> {
     let value: JsonValue =
@@ -19,6 +21,7 @@ pub fn json_str_to_bson_document(json_str: &str) -> std::result::Result<Document
         bson::to_document(&value).map_err(|e| DB3Error::InvalidDocumentBytes(format!("{}", e)))?;
     Ok(bson_document)
 }
+
 pub fn json_str_to_index(json_str: &str, idx: u32) -> std::result::Result<Index, DB3Error> {
     let value: JsonValue =
         serde_json::from_str(json_str).map_err(|e| DB3Error::InvalidJson(format!("{}", e)))?;
@@ -154,53 +157,125 @@ pub fn bson_value_from_proto_value(value: &Value) -> std::result::Result<Bson, D
         Err(DB3Error::InvalidFilterValue("value is none".to_string()))
     }
 }
+
+fn field_filter_from_json_value(
+    filter_doc: &Document,
+) -> std::result::Result<Option<Filter>, DB3Error> {
+    let field = filter_doc.get_str("field").map_err(|e| {
+        DB3Error::InvalidFilterJson(format!("filed is required in filter json for {e}"))
+    })?;
+    let value = match filter_doc.get("value") {
+        Some(v) => filter_value_from_bson_value(v)?,
+        None => {
+            return Err(DB3Error::InvalidFilterJson(
+                "value is required in filter json".to_string(),
+            ));
+        }
+    };
+
+    let op_str = filter_doc
+        .get_str("op")
+        .map_err(|_| DB3Error::InvalidFilterJson("op is required in filter json".to_string()))?;
+    let op = match op_str {
+        "==" => Operator::Equal,
+        ">" => Operator::GreaterThan,
+        "<" => Operator::LessThan,
+        ">=" => Operator::GreaterThanOrEqual,
+        "<=" => Operator::LessThanOrEqual,
+        "!=" => {
+            return Err(DB3Error::InvalidFilterOp(format!(
+                "OP {} un-support currently",
+                op_str
+            )));
+        }
+        _ => {
+            return Err(DB3Error::InvalidFilterOp(format!("Invalid OP {}", op_str)));
+        }
+    };
+
+    Ok(Some(Filter {
+        filter_type: Some(FilterType::FieldFilter(FieldFilter {
+            field: field.to_string(),
+            op: op.into(),
+            value: Some(value),
+        })),
+    }))
+}
+
+fn composite_filter_from_json_value(
+    filters_doc: &Array,
+    op: CompositeOp,
+) -> std::result::Result<Option<Filter>, DB3Error> {
+    if filters_doc.is_empty() {
+        return Err(DB3Error::InvalidFilterJson("filters is empty".to_string()));
+    }
+    let mut filters = vec![];
+    for filter in filters_doc {
+        if let Some(filter_doc) = filter.as_document() {
+            let op_str = filter_doc.get_str("op").map_err(|_| {
+                DB3Error::InvalidFilterJson("op is required in filter json".to_string())
+            })?;
+
+            // only support == in composite filter
+            if op_str != "==" {
+                return Err(DB3Error::InvalidFilterJson(format!(
+                    "{} is not support in composite filter",
+                    op_str
+                )));
+            };
+            if let Ok(Some(filter)) = field_filter_from_json_value(filter_doc) {
+                filters.push(filter);
+            } else {
+                return Err(DB3Error::InvalidFilterJson(
+                    "invalid field filter".to_string(),
+                ));
+            }
+        } else {
+            return Err(DB3Error::InvalidFilterJson("invalid document".to_string()));
+        }
+    }
+
+    Ok(Some(Filter {
+        filter_type: Some(FilterType::CompositeFilter(CompositeFilter {
+            filters,
+            op: op.into(),
+        })),
+    }))
+}
+
 pub fn filter_from_json_value(json_str: &str) -> std::result::Result<Option<Filter>, DB3Error> {
     if json_str.is_empty() {
         Ok(None)
     } else {
         let filter_doc = json_str_to_bson_document(json_str)
             .map_err(|e| DB3Error::InvalidFilterValue(format!("{:?}", e)))?;
-        let field = filter_doc.get_str("field").map_err(|e| {
-            DB3Error::InvalidFilterJson(format!("filed is required in filter json for {e}"))
-        })?;
-        let value = match filter_doc.get("value") {
-            Some(v) => filter_value_from_bson_value(v)?,
-            None => {
-                return Err(DB3Error::InvalidFilterJson(
-                    "value is required in filter json".to_string(),
-                ));
-            }
-        };
 
-        let op_str = filter_doc.get_str("op").map_err(|_| {
-            DB3Error::InvalidFilterJson("op is required in filter json".to_string())
-        })?;
-        let op = match op_str {
-            "==" => Operator::Equal,
-            ">" => Operator::GreaterThan,
-            "<" => Operator::LessThan,
-            ">=" => Operator::GreaterThanOrEqual,
-            "<=" => Operator::LessThanOrEqual,
-            "!=" => {
-                return Err(DB3Error::InvalidFilterOp(format!(
-                    "OP {} un-support currently",
-                    op_str
-                )));
+        if filter_doc.contains_key("field") {
+            field_filter_from_json_value(&filter_doc)
+        } else if filter_doc.contains_key("AND") {
+            if let Ok(filters) = filter_doc.get_array("AND") {
+                composite_filter_from_json_value(filters, CompositeOp::And)
+            } else {
+                Err(DB3Error::InvalidFilterJson(
+                    "filter json is invalid".to_string(),
+                ))
             }
-            _ => {
-                return Err(DB3Error::InvalidFilterOp(format!("Invalid OP {}", op_str)));
+        } else if filter_doc.contains_key("and") {
+            if let Ok(filters) = filter_doc.get_array("and") {
+                composite_filter_from_json_value(filters, CompositeOp::And)
+            } else {
+                Err(DB3Error::InvalidFilterJson(
+                    "filter json is invalid".to_string(),
+                ))
             }
-        };
-
-        Ok(Some(Filter {
-            filter_type: Some(FilterType::FieldFilter(FieldFilter {
-                field: field.to_string(),
-                op: op.into(),
-                value: Some(value),
-            })),
-        }))
+        } else {
+            Err(DB3Error::InvalidFilterJson(
+                "filter json is invalid".to_string(),
+            ))
+        }
     }
 }
+
 pub fn filter_value_from_bson_value(value: &Bson) -> std::result::Result<Value, DB3Error> {
     match value {
         Bson::Boolean(b) => Ok(Value {
@@ -221,6 +296,7 @@ pub fn filter_value_from_bson_value(value: &Bson) -> std::result::Result<Value,
         ))),
     }
 }
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -231,6 +307,7 @@ mod tests {
     use bson::Bson;
     use chrono::Utc;
     use db3_proto::db3_database_proto::index::index_field::{Order, ValueMode};
+
     #[test]
     fn json_str_to_bson_document_ut() {
         let data = r#"
@@ -310,6 +387,7 @@ mod tests {
         assert!(i64_big_value1 < i64_big_value2);
         assert!(i64_big_value2 < i64_max);
     }
+
     #[test]
     fn i32_bson_into_comparison_bytes_ut() {
         let i32_small_value1 = -(0x7F000000 as i32);
@@ -382,7 +460,7 @@ mod tests {
     }
 
     #[test]
-    fn filter_from_json_value_ut() {
+    fn field_filter_from_json_value_ut() {
         let filter = filter_from_json_value("").unwrap();
         assert!(filter.is_none());
 
@@ -442,6 +520,50 @@ mod tests {
         assert!(filter_from_json_value("{}").is_err());
         assert!(filter_from_json_value(r#"{"field": "name"}"#).is_err());
     }
+
+    #[test]
+    fn composite_filter_from_json_value_ut() {
+        let filter = filter_from_json_value(
+            r#"{
+            "and": [
+                {"field": "name", "value": "Bill", "op": "=="},
+                {"field": "age", "value": 44, "op": "=="}
+            ]
+        }"#,
+        )
+        .unwrap()
+        .unwrap();
+        assert_eq!(
+            r#"{"filter_type":{"CompositeFilter":{"op":1,"filters":[{"filter_type":{"FieldFilter":{"field":"name","op":5,"value":{"value_type":{"StringValue":"Bill"}}}}},{"filter_type":{"FieldFilter":{"field":"age","op":5,"value":{"value_type":{"IntegerValue":44}}}}}]}}}"#,
+            serde_json::to_string(&filter).unwrap()
+        );
+
+        assert!(filter_from_json_value(
+            r#"{
+            "and": []
+        }"#
+        )
+        .is_err());
+        assert!(filter_from_json_value(
+            r#"{
+            "or": [
+                {"field": "name", "value": "Bill", "op": "=="},
+                {"field": "age", "value": 44, "op": "=="}
+            ]
+        }"#
+        )
+        .is_err());
+        assert!(filter_from_json_value(
+            r#"{
+            "and": [
+                {"field": "name", "value": "Bill", "op": "=="},
+                {"field": "age", "value": 44, "op": ">="}
+            ]
+        }"#
+        )
+        .is_err());
+    }
+
     #[test]
     fn bson_value_from_proto_value_ut() {
         assert!(bson_value_from_proto_value(&Value { value_type: None }).is_err());
@@ -501,6 +623,7 @@ mod tests {
             .unwrap()
         );
     }
+
     #[test]
     fn json_str_to_index_ut() {
         let json_str = r#"{"name":"idx1","fields":[{"field_path":"name","value_mode":{"Order":1}}, {"field_path":"age","value_mode":{"Order":1}}]}"#;
@@ -524,12 +647,14 @@ mod tests {
         };
         assert_eq!(expect, index);
     }
+
     #[test]
     fn json_str_to_index_wrong_path_1_ut() {
         let json_str = r#"{"fields":[{"field_path":"name","value_mode":{"Order":1}}, {"field_path":"age","value_mode":{"Order":1}}]}"#;
         let res = json_str_to_index(json_str, 1);
         assert!(res.is_err());
     }
+
     #[test]
     fn json_str_to_index_wrong_path_2_ut() {
         let json_str = r#"{"name": "idx1""#;
diff --git a/src/cmd/src/command.rs b/src/cmd/src/command.rs
index 00597ed6..451d16e4 100644
--- a/src/cmd/src/command.rs
+++ b/src/cmd/src/command.rs
@@ -513,11 +513,11 @@ impl DB3ClientCommand {
                     );
                 }
                 // we currently only support index filter single field
-                for index in index_vec.iter() {
-                    if index.fields.len() != 1 {
-                        return Err("fail to create collection with multi-key index".to_string());
-                    }
-                }
+                // for index in index_vec.iter() {
+                //     if index.fields.len() != 1 {
+                //         return Err("fail to create collection with multi-key index".to_string());
+                //     }
+                // }
                 let collection = CollectionMutation {
                     index: index_vec.to_owned(),
                     collection_name: name.to_string(),
diff --git a/src/node/src/command.rs b/src/node/src/command.rs
index 69fdea6d..180b59be 100644
--- a/src/node/src/command.rs
+++ b/src/node/src/command.rs
@@ -625,6 +625,11 @@ mod tests {
                     .to_string(),
                 r#"{"name":"idx2","fields":[{"field_path":"age","value_mode":{"Order":1}}]}"#
                     .to_string(),
+                r#"{"name":"idx_name_age","fields":[
+                    {"field_path":"name","value_mode":{"Order":1}},
+                    {"field_path":"age","value_mode":{"Order":1}}
+                ]}"#
+                .to_string(),
             ],
         };
 
@@ -740,6 +745,22 @@ mod tests {
                 r#"{"field": "name", "value": "John Doe", "op": ">"}"#,
                 vec![r#""name": String("Mike")"#],
             ),
+            (
+                r#"{"and":
+                [
+                    {"field": "name", "value": "Bill", "op": "=="},
+                    {"field": "age", "value": 44, "op": "=="}
+                ]}"#,
+                vec![r#""name": String("Bill")"#],
+            ),
+            (
+                r#"{"and":
+                [
+                    {"field": "name", "value": "Bill", "op": "=="},
+                    {"field": "age", "value": 46, "op": "=="}
+                ]}"#,
+                vec![],
+            ),
         ] {
             let cmd = DB3ClientCommand::ShowDocument {
                 addr: addr.clone(),
@@ -747,7 +768,8 @@ mod tests {
                 filter: filter.to_string(),
                 limit: -1,
             };
-            if let Ok(table) = cmd.execute(&mut ctx).await {
+            let res = cmd.execute(&mut ctx).await;
+            if let Ok(table) = res {
                 assert_eq!(exp.len(), table.len());
                 for i in 0..exp.len() {
                     assert!(
@@ -764,7 +786,7 @@ mod tests {
                     );
                 }
             } else {
-                assert!(false)
+                assert!(false, "{:?}", res);
             }
         }
         // run show document --filter = '{"field": "age", "value": 44, "op": "=="}'
diff --git a/src/proto/proto/db3_database.proto b/src/proto/proto/db3_database.proto
index fdef0127..14e7a9c6 100644
--- a/src/proto/proto/db3_database.proto
+++ b/src/proto/proto/db3_database.proto
@@ -149,8 +149,7 @@ message StructuredQuery {
       // A filter on a document field.
       FieldFilter field_filter = 1;
       // A composite filter.
-      // TODO: Support in the future P2
-      // CompositeFilter composite_filter = 2;
+      CompositeFilter composite_filter = 2;
 
       // A filter that takes exactly one argument.
       // TODO: Support in the future P1
@@ -158,6 +157,31 @@ message StructuredQuery {
     }
   }
 
+  // A filter that merges multiple other filters using the given operator.
+  message CompositeFilter {
+    // A composite filter operator.
+    enum Operator {
+      // Unspecified. This value must not be used.
+      OPERATOR_UNSPECIFIED = 0;
+
+      // Documents are required to satisfy all of the combined filters.
+      AND = 1;
+
+      // Documents are required to satisfy at least one of the combined filters.
+      // TODO: support or in the P1
+      // OR = 2;
+    }
+
+    // The operator for combining multiple filters.
+    Operator op = 1;
+
+    // The list of filters to combine.
+    //
+    // Requires:
+    //
+    // * At least one filter is present.
+    repeated Filter filters = 2;
+  }
   // A message that can hold any of the supported value types.
   message Value {
     // Must have a value set.
diff --git a/src/storage/src/db_store.rs b/src/storage/src/db_store.rs
index 2d7ad12c..fbdcabf7 100644
--- a/src/storage/src/db_store.rs
+++ b/src/storage/src/db_store.rs
@@ -24,6 +24,8 @@ use db3_crypto::{
     id::IndexId, id::TxId,
 };
 use db3_error::{DB3Error, Result};
+use db3_proto::db3_database_proto::index::IndexField;
+use db3_proto::db3_database_proto::structured_query::composite_filter::Operator as CompositeOp;
 use db3_proto::db3_database_proto::structured_query::field_filter::Operator;
 use db3_proto::db3_database_proto::structured_query::filter::FilterType;
 use db3_proto::db3_database_proto::{Collection, Database, Document, Index, StructuredQuery};
@@ -168,16 +170,6 @@ impl DbStore {
         );
         let (db, mut ops) =
             Self::new_database(&dbid, sender, tx, mutation, block_id, mutation_id, desc);
-        for c in &db.collections {
-            for index in &c.index_list {
-                if index.fields.len() > 1 {
-                    return Err(DB3Error::ApplyDatabaseError(format!(
-                        "currently not support multi-key index, index fields should be less than 1, but got {}",
-                        index.fields.len()
-                    )));
-                }
-            }
-        }
         let (batches, data_in_bytes) =
             Self::encode_database(dbid, &db, sender, block_id, mutation_id, true)?;
         ops.update_data_size(data_in_bytes as u64);
@@ -686,6 +678,24 @@ impl DbStore {
         field_index_map
     }
 
+    /// check if the fields_names is a prefix of field_indexes
+    /// e.g. fields_names = ["a", "b"]
+    ///     field_indexes = ["a", "b", "c", "d"]
+    fn index_start_with_fields(
+        fields_names: &Vec<String>,
+        field_indexes: &Vec<IndexField>,
+    ) -> bool {
+        if fields_names.len() > field_indexes.len() {
+            return false;
+        }
+
+        for i in 0..fields_names.len() {
+            if fields_names[i] != field_indexes[i].field_path {
+                return false;
+            }
+        }
+        true
+    }
     /// run a query to fetch target documents from given database and collection
     pub fn run_query(
         db: Pin<&Merk>,
@@ -748,6 +758,104 @@ impl DbStore {
                                 )?;
                                 Self::execute_query(db, &range, limit)
                             }
+                            Some(FilterType::CompositeFilter(composite_filter)) => {
+                                match CompositeOp::from_i32(composite_filter.op) {
+                                    Some(CompositeOp::And) => {
+                                        let mut filter_names = vec![];
+                                        let mut keys = vec![];
+
+                                        for filter in composite_filter.filters.iter() {
+                                            match &filter.filter_type {
+                                                Some(FilterType::FieldFilter(field_filter)) => {
+                                                    match &field_filter.value {
+                                                        Some(value) => {
+                                                            if field_filter.op
+                                                                != Operator::Equal as i32
+                                                            {
+                                                                return Err(DB3Error::InvalidFilterValue(
+                                                                    "CompositeOp And only support Equal".to_string(),
+                                                                ));
+                                                            }
+                                                            filter_names.push(
+                                                                field_filter.field.to_string(),
+                                                            );
+                                                            keys.push(bson_util::bson_into_comparison_bytes(
+                                                                &bson_util::bson_value_from_proto_value(value)?,
+                                                            )?);
+                                                        }
+                                                        None => {
+                                                            return Err(DB3Error::InvalidFilterValue(
+                                                                "None field filter value un-support".to_string(),
+                                                            ));
+                                                        }
+                                                    }
+                                                }
+                                                _ => {
+                                                    return Err(DB3Error::InvalidFilterValue(
+                                                        "CompositeOp And only support FieldFilter"
+                                                            .to_string(),
+                                                    ));
+                                                }
+                                            }
+                                        }
+
+                                        if filter_names.is_empty() {
+                                            return Err(DB3Error::InvalidFilterValue(
+                                                "CompositeOp can't support empty filters"
+                                                    .to_string(),
+                                            ));
+                                        }
+
+                                        let index = match field_index_map.get(&filter_names[0]) {
+                                            Some(index_list) => {
+                                                match index_list.iter().find_or_first(|i| {
+                                                    Self::index_start_with_fields(
+                                                        &filter_names,
+                                                        &i.fields,
+                                                    )
+                                                }) {
+                                                    Some(index_match) => index_match,
+                                                    None => {
+                                                        return Err(
+                                                            DB3Error::IndexNotFoundForFiledFilter(
+                                                                filter_names[0].to_string(),
+                                                            ),
+                                                        );
+                                                    }
+                                                }
+                                            }
+                                            None => {
+                                                return Err(DB3Error::IndexNotFoundForFiledFilter(
+                                                    filter_names[0].to_string(),
+                                                ));
+                                            }
+                                        };
+                                        debug!("filter names: {:?}", filter_names);
+                                        debug!("keys: {:?}", keys);
+                                        debug!("index: {:?}", index);
+                                        let mut key = vec![];
+
+                                        for k in keys {
+                                            key.extend_from_slice(k.as_slice());
+                                        }
+
+                                        let range = Self::generate_range_with_single_field_filter(
+                                            &collection_id,
+                                            index,
+                                            &key,
+                                            Some(Operator::Equal),
+                                        )?;
+                                        debug!("range: {:?}", range);
+                                        Self::execute_query(db, &range, limit)
+                                    }
+                                    _ => {
+                                        return Err(DB3Error::InvalidFilterType(format!(
+                                            "Invalid composite op {:?}",
+                                            CompositeOp::from_i32(composite_filter.op)
+                                        )));
+                                    }
+                                }
+                            }
                             None => {
                                 return Err(DB3Error::InvalidFilterType(
                                     "None filter type unsupport".to_string(),
@@ -1095,7 +1203,7 @@ mod tests {
     use db3_crypto::key_derive;
     use db3_crypto::signature_scheme::SignatureScheme;
     use db3_proto::db3_database_proto::structured_query::{
-        value::ValueType, FieldFilter, Filter, Limit, Projection, Value,
+        value::ValueType, CompositeFilter, FieldFilter, Filter, Limit, Projection, Value,
     };
     use db3_proto::db3_database_proto::{
         index::index_field::{Order, ValueMode},
@@ -1242,6 +1350,7 @@ mod tests {
         println!("{json_data}");
         dm
     }
+
     fn build_database_mutation_with_multi_key_index(
         addr: &DB3Address,
         collection_name: &str,
@@ -1391,7 +1500,289 @@ mod tests {
     }
 
     #[test]
-    fn db_store_run_query_test() {
+    fn db_store_run_composite_filter_query_test() {
+        let tmp_dir_path =
+            TempDir::new("db_store_run_composite_filter_query_test").expect("create temp dir");
+        let merk = Merk::open(tmp_dir_path).unwrap();
+        let mut db = Box::pin(merk);
+        let collection_name = "db_store_run_composite_filter_query_test".to_string();
+        let block_id: u64 = 1001;
+
+        // create DB Test
+        let addr = gen_address();
+        let db_id = DbId::try_from((&addr, 1)).unwrap();
+        let db_mutation = build_database_mutation_with_multi_key_index(
+            &db_id.address(),
+            collection_name.as_str(),
+        );
+        let db_m: Pin<&mut Merk> = Pin::as_mut(&mut db);
+        let result =
+            DbStore::apply_mutation(db_m, &addr, 1, &TxId::zero(), &db_mutation, block_id, 1);
+        assert!(result.is_ok());
+
+        // add 4 documents into collection
+        let db_mutation = build_add_document_mutation(
+            db_id.address(),
+            collection_name.as_str(),
+            vec![
+                r#"
+        {
+            "name": "John Doe",
+            "age": 43,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "Mike",
+            "age": 44,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "Bill",
+            "age": 44,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "Bill",
+            "age": 45,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+            ],
+        );
+
+        // add document test
+        let db_m: Pin<&mut Merk> = Pin::as_mut(&mut db);
+        let (_, ids) =
+            DbStore::add_document(db_m, &addr, &TxId::zero(), &db_mutation, block_id, 2).unwrap();
+        assert_eq!(4, ids.len());
+
+        // test query with composite filter
+        for (name, age, exp) in [
+            // Select * from collection where name = "Bill" and age = 44
+            ("Bill", 44, vec![("Bill", 44)]),
+            // Select * from collection where name = "Bill" and age = 45
+            ("Bill", 45, vec![("Bill", 45)]),
+            // Select * from collection where name = "Bill" and age = 46
+            ("Bill", 46, vec![]),
+            // Select * from collection where name = "Mike" and age = 44
+            ("Mike", 44, vec![("Mike", 44)]),
+        ] {
+            let query = StructuredQuery {
+                collection_name: collection_name.to_string(),
+                select: Some(Projection { fields: vec![] }),
+                r#where: Some(Filter {
+                    filter_type: Some(FilterType::CompositeFilter(CompositeFilter {
+                        filters: vec![
+                            Filter {
+                                filter_type: Some(FilterType::FieldFilter(FieldFilter {
+                                    field: "name".to_string(),
+                                    op: Operator::Equal.into(),
+                                    value: Some(Value {
+                                        value_type: Some(ValueType::StringValue(name.to_string())),
+                                    }),
+                                })),
+                            },
+                            Filter {
+                                filter_type: Some(FilterType::FieldFilter(FieldFilter {
+                                    field: "age".to_string(),
+                                    op: Operator::Equal.into(),
+                                    value: Some(Value {
+                                        value_type: Some(ValueType::IntegerValue(age as i64)),
+                                    }),
+                                })),
+                            },
+                        ],
+                        op: CompositeOp::And.into(),
+                    })),
+                }),
+                limit: None,
+            };
+            let docs = DbStore::run_query(db.as_ref(), &db_id, &query).unwrap();
+            assert_eq!(exp.len(), docs.len(), "run query fail for {:?}", query);
+            for i in 0..exp.len() {
+                let document = bson_util::bytes_to_bson_document(docs[i].doc.clone()).unwrap();
+                assert_eq!(
+                    exp[i].0,
+                    document.get_str("name").unwrap(),
+                    "run query fail for {:?}",
+                    query
+                );
+                assert_eq!(
+                    exp[i].1,
+                    document.get_i64("age").unwrap(),
+                    "run query fail for {:?}",
+                    query
+                );
+            }
+        }
+    }
+    #[test]
+    fn db_store_run_composite_filter_query_wrong_path_test() {
+        let tmp_dir_path = TempDir::new("db_store_run_composite_filter_query_wrong_path_test")
+            .expect("create temp dir");
+        let merk = Merk::open(tmp_dir_path).unwrap();
+        let mut db = Box::pin(merk);
+        let collection_name = "db_store_run_composite_filter_query_wrong_path_test".to_string();
+        let block_id: u64 = 1001;
+
+        // create DB Test
+        let addr = gen_address();
+        let db_id = DbId::try_from((&addr, 1)).unwrap();
+        let db_mutation = build_database_mutation_with_multi_key_index(
+            &db_id.address(),
+            collection_name.as_str(),
+        );
+        let db_m: Pin<&mut Merk> = Pin::as_mut(&mut db);
+        let result =
+            DbStore::apply_mutation(db_m, &addr, 1, &TxId::zero(), &db_mutation, block_id, 1);
+        assert!(result.is_ok());
+
+        // add 4 documents into collection
+        let db_mutation = build_add_document_mutation(
+            db_id.address(),
+            collection_name.as_str(),
+            vec![
+                r#"
+        {
+            "name": "John Doe",
+            "age": 43,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "Mike",
+            "age": 44,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "Bill",
+            "age": 44,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "Bill",
+            "age": 45,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
+        }"#
+                .to_string(),
+            ],
+        );
+
+        // add document test
+        let db_m: Pin<&mut Merk> = Pin::as_mut(&mut db);
+        let (_, ids) =
+            DbStore::add_document(db_m, &addr, &TxId::zero(), &db_mutation, block_id, 2).unwrap();
+        assert_eq!(4, ids.len());
+
+        // test query with composite filter
+        for (name, name_op, age, age_op, composite_op, is_ok) in [
+            // Select * from collection where name = "Bill" and age = 44
+            (
+                "Bill",
+                Operator::Equal,
+                44,
+                Operator::Equal,
+                CompositeOp::And,
+                true,
+            ),
+            (
+                "Bill",
+                Operator::Equal,
+                44,
+                Operator::Equal,
+                CompositeOp::Unspecified,
+                false,
+            ),
+            (
+                "Bill",
+                Operator::LessThan,
+                44,
+                Operator::Equal,
+                CompositeOp::And,
+                false,
+            ),
+            (
+                "Bill",
+                Operator::Equal,
+                44,
+                Operator::LessThan,
+                CompositeOp::And,
+                false,
+            ),
+        ] {
+            let query = StructuredQuery {
+                collection_name: collection_name.to_string(),
+                select: Some(Projection { fields: vec![] }),
+                r#where: Some(Filter {
+                    filter_type: Some(FilterType::CompositeFilter(CompositeFilter {
+                        filters: vec![
+                            Filter {
+                                filter_type: Some(FilterType::FieldFilter(FieldFilter {
+                                    field: "name".to_string(),
+                                    op: name_op.into(),
+                                    value: Some(Value {
+                                        value_type: Some(ValueType::StringValue(name.to_string())),
+                                    }),
+                                })),
+                            },
+                            Filter {
+                                filter_type: Some(FilterType::FieldFilter(FieldFilter {
+                                    field: "age".to_string(),
+                                    op: age_op.into(),
+                                    value: Some(Value {
+                                        value_type: Some(ValueType::IntegerValue(age as i64)),
+                                    }),
+                                })),
+                            },
+                        ],
+                        op: composite_op.into(),
+                    })),
+                }),
+                limit: None,
+            };
+            let res = DbStore::run_query(db.as_ref(), &db_id, &query);
+            assert_eq!(is_ok, res.is_ok());
+            println!("res: {:?}", res);
+        }
+    }
+
+    #[test]
+    fn db_store_run_simple_query_test() {
         let tmp_dir_path = TempDir::new("db_store_run_query_test").expect("create temp dir");
         let merk = Merk::open(tmp_dir_path).unwrap();
         let mut db = Box::pin(merk);
@@ -2040,20 +2431,22 @@ mod tests {
             format!("{:?}", res.err().unwrap())
         );
     }
+
     #[test]
-    fn db_store_create_database_wrong_path() {
-        let tmp_dir_path = TempDir::new("db_store_create_database_test").expect("create temp dir");
+    fn db_store_create_database_happy_path() {
+        let tmp_dir_path =
+            TempDir::new("db_store_create_database_happy_path").expect("create temp dir");
         let addr = gen_address();
         let merk = Merk::open(tmp_dir_path).unwrap();
         let mut db = Box::pin(merk);
-        let collection_name = "db_store_create_database_test".to_string();
+        let collection_name = "db_store_create_database_happy_path".to_string();
         let db_mutation =
             build_database_mutation_with_multi_key_index(&addr, collection_name.as_str());
         let db_m: Pin<&mut Merk> = Pin::as_mut(&mut db);
 
         // create DB wrong path
         let result = DbStore::apply_mutation(db_m, &addr, 1, &TxId::zero(), &db_mutation, 1000, 1);
-        assert!(result.is_err());
+        assert!(result.is_ok());
     }
 
     #[test]


Test patch: