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

Issue:
WIP: Propose index encode solution
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->

- Would you like to implement this feature? [y/n]


### Goal

- index key encode into/decode from bytes
- index bytes can be sorted
- index bytes order should be consistent with the original value order
- support Null encode
- support Emptry String encode
- define the order between types. Reference: https://www.mongodb.com/docs/manual/reference/bson-type-comparison-order/
- consider encode/decode performance and memory usage

### Reference
https://www.mongodb.com/docs/manual/reference/bson-type-comparison-order/

### Context

We currently support simple document index to converge the very basic use case. e.g (String index, Int64, Int32, Boolean)

In this issue, we aim to expand our support to include additional features such as Null value, empty String, null String, multi-index scenario, etc"

With respect to that, we need to review our index mechanism and then improve or redesign it if necessary.

## 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:
@jingchen2222  Has this issue been resolved?
not yep

On Fri, Apr 14, 2023 at 8:42 PM imotai ***@***.***> wrote:

> @jingchen2222 <https://github.com/jingchen2222> Has this issue been
> resolved?
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/dbpunk-labs/db3/issues/321#issuecomment-1509488788>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/ABP75ZND4PAPBGYD55DHFO3XBIKIRANCNFSM6AAAAAAU764KGA>
> .
> You are receiving this because you were mentioned.Message ID:
> ***@***.***>
>

> not yep
> […](#)
> On Fri, Apr 14, 2023 at 8:42 PM imotai ***@***.***> wrote: @jingchen2222 <https://github.com/jingchen2222> Has this issue been resolved? — Reply to this email directly, view it on GitHub <[#321 (comment)](https://github.com/dbpunk-labs/db3/issues/321#issuecomment-1509488788)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABP75ZND4PAPBGYD55DHFO3XBIKIRANCNFSM6AAAAAAU764KGA> . You are receiving this because you were mentioned.Message ID: ***@***.***>

We just solved the key order problem

Fix patch:
diff --git a/src/base/src/bson_util.rs b/src/base/src/bson_util.rs
index 6e0818e7..63353099 100644
--- a/src/base/src/bson_util.rs
+++ b/src/base/src/bson_util.rs
@@ -1,7 +1,6 @@
 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;
@@ -67,82 +66,6 @@ pub fn bson_document_into_bytes(doc: &Document) -> Vec<u8> {
     row_doc.into_bytes()
 }
 
-fn keep_order_i32(input: i32) -> u32 {
-    match input < 0 {
-        true => {
-            if input == i32::MIN {
-                0
-            } else {
-                let new_input = input as u32;
-                (new_input & 0x7fffffff) as u32
-            }
-        }
-        false => {
-            let new_input = input as u32;
-            (new_input | 0x80000000) as u32
-        }
-    }
-}
-
-fn keep_order_i64(input: i64) -> u64 {
-    match input < 0 {
-        true => {
-            if input == i64::MIN {
-                0
-            } else {
-                let new_input = input as u64;
-                (new_input & 0x7fffffffffffffff) as u64
-            }
-        }
-        false => {
-            let new_input = input as u64;
-            (new_input | 0x8000000000000000) as u64
-        }
-    }
-}
-
-/// convert bson value to bytes for key comparation
-pub fn bson_into_comparison_bytes(value: &Bson) -> std::result::Result<Vec<u8>, DB3Error> {
-    let mut data: Vec<u8> = Vec::new();
-    match value {
-        Bson::Null => {
-            // TODO(chanjing): suuport NULL encode bytes in the future. P1
-            Err(DB3Error::DocumentDecodeError(
-                "null type is not supported".to_string(),
-            ))
-        }
-        Bson::Boolean(b) => {
-            data.write_u8(*b as u8)
-                .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
-            Ok(data)
-        }
-        Bson::Int64(n) => {
-            data.write_u64::<BigEndian>(keep_order_i64(*n))
-                .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
-            Ok(data)
-        }
-        Bson::Int32(n) => {
-            data.write_u32::<BigEndian>(keep_order_i32(*n))
-                .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
-            Ok(data)
-        }
-        // TODO: add \0 as the end of string.
-        Bson::String(s) => {
-            data.extend_from_slice(s.as_bytes());
-            Ok(data)
-        }
-        Bson::DateTime(dt) => {
-            let value: u64 = keep_order_i64(dt.timestamp_millis());
-            data.write_u64::<BigEndian>(value)
-                .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
-            Ok(data)
-        }
-        _ => Err(DB3Error::DocumentDecodeError(
-            "value type is not supported".to_string(),
-        )),
-    }
-}
-
 pub fn bson_value_from_proto_value(value: &Value) -> std::result::Result<Bson, DB3Error> {
     if let Some(value_type) = &value.value_type {
         match value_type {
@@ -301,11 +224,9 @@ pub fn filter_value_from_bson_value(value: &Bson) -> std::result::Result<Value,
 mod tests {
     use super::*;
     use crate::bson_util::{
-        bson_document_into_bytes, bson_into_comparison_bytes, bytes_to_bson_document,
-        json_str_to_bson_document,
+        bson_document_into_bytes, bytes_to_bson_document, json_str_to_bson_document,
     };
     use bson::Bson;
-    use chrono::Utc;
     use db3_proto::db3_database_proto::index::index_field::{Order, ValueMode};
 
     #[test]
@@ -344,121 +265,6 @@ mod tests {
         )
     }
 
-    #[test]
-    fn i64_bson_into_comparison_bytes_ut() {
-        let i64_neg_2 = bson_into_comparison_bytes(&Bson::Int64(-2)).unwrap();
-        let i64_neg_1 = bson_into_comparison_bytes(&Bson::Int64(-1)).unwrap();
-        let i64_small_value1 = -(0x7F00000000000000 as i64);
-        let i64_small_1 = bson_into_comparison_bytes(&Bson::Int64(i64_small_value1)).unwrap();
-        let i64_small_value2 = -(0x7000000000000000 as i64);
-        let i64_small_2 = bson_into_comparison_bytes(&Bson::Int64(i64_small_value2)).unwrap();
-        let i64_0 = bson_into_comparison_bytes(&Bson::Int64(0)).unwrap();
-        let i64_1 = bson_into_comparison_bytes(&Bson::Int64(1)).unwrap();
-        let i64_big_value1 =
-            bson_into_comparison_bytes(&Bson::Int64(0x7000000000000000 as i64)).unwrap();
-        let i64_big_value2 =
-            bson_into_comparison_bytes(&Bson::Int64(0x7F00000000000000 as i64)).unwrap();
-        let i64_max = bson_into_comparison_bytes(&Bson::Int64(i64::MAX)).unwrap();
-        let i64_min = bson_into_comparison_bytes(&Bson::Int64(i64::MIN)).unwrap();
-        println!("i64_min: {:?}", i64_min);
-        println!("{} i64_small_value1: {:?}", i64_small_value1, i64_small_1);
-        println!("{} i64_small_value2: {:?}", i64_small_value2, i64_small_2);
-        println!("i64_-2: {:?}", i64_neg_2);
-        println!("i64_-1: {:?}", i64_neg_1);
-        println!("i64_0: {:?}", i64_0);
-        println!("i64_1: {:?}", i64_1);
-        println!(
-            "{} i64_big_value1: {:?}",
-            0x7000000000000000 as i64, i64_big_value1
-        );
-        println!(
-            "{} i64_big_value2: {:?}",
-            0x7F00000000000000 as i64, i64_big_value2
-        );
-        println!("i64_max: {:?}", i64_max);
-
-        assert!(i64_min < i64_small_1);
-        assert!(i64_small_1 < i64_small_2);
-        assert!(i64_small_2 < i64_neg_1);
-        assert!(i64_neg_2 < i64_1);
-        assert!(i64_neg_1 < i64_0);
-        assert!(i64_0 < i64_1);
-        assert!(i64_1 < i64_big_value1);
-        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);
-        let i32_small_1 = bson_into_comparison_bytes(&Bson::Int32(i32_small_value1)).unwrap();
-        let i32_small_value2 = -(0x70000000 as i32);
-        let i32_small_2 = bson_into_comparison_bytes(&Bson::Int32(i32_small_value2)).unwrap();
-        let i32_neg_2 = bson_into_comparison_bytes(&Bson::Int32(-2)).unwrap();
-        let i32_neg_1 = bson_into_comparison_bytes(&Bson::Int32(-1)).unwrap();
-        let i32_0 = bson_into_comparison_bytes(&Bson::Int32(0)).unwrap();
-        let i32_1 = bson_into_comparison_bytes(&Bson::Int32(1)).unwrap();
-        let i32_big_value1 = bson_into_comparison_bytes(&Bson::Int32(0x70000000 as i32)).unwrap();
-        let i32_big_value2 = bson_into_comparison_bytes(&Bson::Int32(0x7F000000 as i32)).unwrap();
-        let i32_max = bson_into_comparison_bytes(&Bson::Int32(i32::MAX)).unwrap();
-        let i32_min = bson_into_comparison_bytes(&Bson::Int32(i32::MIN)).unwrap();
-
-        println!("i32_min: {:?}", i32_min);
-        println!("{} i32_small_1: {:?}", i32_small_value1, i32_small_1);
-        println!("{} i32_small_2: {:?}", i32_small_value2, i32_small_2);
-        println!("i32_-2: {:?}", i32_neg_2);
-        println!("i32_-1: {:?}", i32_neg_1);
-        println!("i32_0: {:?}", i32_0);
-        println!("i32_1: {:?}", i32_1);
-        println!("{} i32_big_value1: {:?}", 0x70000000 as i32, i32_big_value1);
-        println!("{} i32_big_value2: {:?}", 0x7F000000 as i32, i32_big_value2);
-        println!("i32_max: {:?}", i32_max);
-
-        assert!(i32_min < i32_small_1);
-        assert!(i32_small_1 < i32_small_2);
-        assert!(i32_small_2 < i32_neg_2);
-        assert!(i32_neg_2 < i32_neg_1);
-        assert!(i32_neg_1 < i32_0);
-        assert!(i32_0 < i32_1);
-        assert!(i32_1 < i32_big_value1);
-        assert!(i32_big_value1 < i32_big_value2);
-        assert!(i32_big_value2 < i32_max);
-    }
-
-    #[test]
-    fn string_bson_into_comparison_bytes_ut() {
-        let empty_str = bson_into_comparison_bytes(&Bson::String("".to_string())).unwrap();
-        let a_str = bson_into_comparison_bytes(&Bson::String("a".to_string())).unwrap();
-        let z_str = bson_into_comparison_bytes(&Bson::String("z".to_string())).unwrap();
-        let a_long_str = bson_into_comparison_bytes(&Bson::String("abcdefg".to_string())).unwrap();
-        assert!(empty_str < a_str);
-        assert!(a_str < z_str);
-        assert!(a_long_str < z_str);
-    }
-
-    #[test]
-    fn datetime_bson_into_comparison_bytes_ut() {
-        let now_ts = Utc::now().timestamp_millis();
-        let now = bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(now_ts)))
-            .unwrap();
-        let now_minus_one =
-            bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(now_ts - 1)))
-                .unwrap();
-        let now_plus_one =
-            bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(now_ts + 1)))
-                .unwrap();
-        let zero_ts =
-            bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(0))).unwrap();
-        let min_ts = bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::MIN)).unwrap();
-        let max_ts = bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::MAX)).unwrap();
-
-        assert!(min_ts < zero_ts);
-        assert!(zero_ts < now_minus_one);
-        assert!(now_minus_one < now);
-        assert!(now < now_plus_one);
-        assert!(now_plus_one < max_ts);
-    }
-
     #[test]
     fn field_filter_from_json_value_ut() {
         let filter = filter_from_json_value("").unwrap();
diff --git a/src/crypto/Cargo.toml b/src/crypto/Cargo.toml
index 6a08a4e6..c1c91ce1 100644
--- a/src/crypto/Cargo.toml
+++ b/src/crypto/Cargo.toml
@@ -28,6 +28,7 @@ serde-name = "0.2.1"
 thiserror = "1.0.34"
 #serde_bytes = "0.11.7"
 #serde_json = "1.0.88"
+bson = "2.5.0"
 serde_json = { version = "1.0.64", default-features = false }
 serde_with = "2.1.0"
 serde_repr = "0.1"
@@ -41,6 +42,6 @@ byteorder = "1.4.3"
 rust_secp256k1 = { version = "0.24.0", package = "secp256k1", features = ["bitcoin_hashes"] }
 bip32 = "0.5.0"
 ethers = { workspace = true }
+chrono = "0.4.22"
 [dev-dependencies]
-tiny-bip39 = "1.0.0"
-
+tiny-bip39 = "1.0.0"
\ No newline at end of file
diff --git a/src/crypto/src/id.rs b/src/crypto/src/id.rs
index 44187fac..524bc6ee 100644
--- a/src/crypto/src/id.rs
+++ b/src/crypto/src/id.rs
@@ -17,13 +17,15 @@
 
 use crate::db3_address::{DB3Address, DB3_ADDRESS_LENGTH};
 use base64ct::Encoding as _;
-use byteorder::ReadBytesExt;
-use byteorder::{BigEndian, WriteBytesExt};
+use bson::Bson;
+use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
+use bytes::Buf;
 use db3_error::{DB3Error, Result};
 use fastcrypto::hash::{HashFunction, Sha3_256};
 use rust_secp256k1::hashes::{sha256, Hash};
 use rust_secp256k1::ThirtyTwoByteHash;
 use std::fmt;
+use std::io::Cursor;
 
 // it's ethereum compatiable account id
 #[derive(Eq, Default, PartialEq, Ord, PartialOrd, Copy, Clone)]
@@ -153,6 +155,9 @@ pub const OP_ENTRY_ID_LENGTH: usize = 12;
 pub const DOCUMENT_ID_TYPE_ID: i8 = 1;
 pub const INDEX_ID_TYPE_ID: i8 = 2;
 
+/// FieldTypeId := 1 bytes
+pub const FIELD_TYPE_ID_LENGTH: usize = 1;
+
 #[derive(Eq, Default, PartialEq, Ord, PartialOrd, Copy, Clone, Debug)]
 pub struct OpEntryId {
     data: [u8; OP_ENTRY_ID_LENGTH],
@@ -318,6 +323,265 @@ impl fmt::Display for DocumentId {
         write!(f, "DOC|{}|{}", collection_id, document_entry_id)
     }
 }
+pub enum FieldTypeId {
+    Null = 0x01,
+    Bool = 0x02,
+    I32 = 0x03,
+    I64 = 0x04,
+    F32 = 0x05,
+    F64 = 0x06,
+    DateTime = 0x07,
+    String = 0x0a,
+}
+#[derive(Eq, Default, PartialEq, Ord, PartialOrd, Clone, Debug)]
+pub struct FieldKey {
+    data: Vec<u8>,
+}
+impl FieldKey {
+    fn encode_i32(input: i32) -> u32 {
+        match input < 0 {
+            true => {
+                if input == i32::MIN {
+                    0
+                } else {
+                    let new_input = input as u32;
+                    (new_input & 0x7fffffff) as u32
+                }
+            }
+            false => {
+                let new_input = input as u32;
+                (new_input | 0x80000000) as u32
+            }
+        }
+    }
+    fn decode_u32(input: u32) -> i32 {
+        if input == 0 {
+            i32::MIN
+        } else {
+            if input as i32 > 0 {
+                (input | 0x80000000) as i32
+            } else {
+                (input & 0x7fffffff) as i32
+            }
+        }
+    }
+
+    fn encode_i64(input: i64) -> u64 {
+        match input < 0 {
+            true => {
+                if input == i64::MIN {
+                    0
+                } else {
+                    let new_input = input as u64;
+                    (new_input & 0x7fffffffffffffff) as u64
+                }
+            }
+            false => {
+                let new_input = input as u64;
+                (new_input | 0x8000000000000000) as u64
+            }
+        }
+    }
+    fn decode_u64(input: u64) -> i64 {
+        if input == 0 {
+            i64::MIN
+        } else {
+            if input as i64 > 0 {
+                (input | 0x8000000000000000) as i64
+            } else {
+                (input & 0x7fffffffffffffff) as i64
+            }
+        }
+    }
+
+    fn add(&mut self, field: &Option<Bson>) -> std::result::Result<(), DB3Error> {
+        match field {
+            None => {
+                self.data
+                    .write_u8(FieldTypeId::Null as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+            }
+
+            Some(Bson::Boolean(b)) => {
+                self.data
+                    .write_u8(FieldTypeId::Bool as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                self.data
+                    .write_u8(*b as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+            }
+            Some(Bson::Int64(n)) => {
+                self.data
+                    .write_u8(FieldTypeId::I64 as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                self.data
+                    .write_u64::<BigEndian>(Self::encode_i64(*n))
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+            }
+            Some(Bson::Int32(n)) => {
+                self.data
+                    .write_u8(FieldTypeId::I32 as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                self.data
+                    .write_u32::<BigEndian>(Self::encode_i32(*n))
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+            }
+            // TODO: add \0 as the end of string.
+            Some(Bson::String(s)) => {
+                self.data
+                    .write_u8(FieldTypeId::String as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                self.data.extend_from_slice(s.as_bytes());
+                self.data.extend_from_slice(&[0]);
+            }
+            Some(Bson::DateTime(dt)) => {
+                self.data
+                    .write_u8(FieldTypeId::DateTime as u8)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                let value: u64 = Self::encode_i64(dt.timestamp_millis());
+                self.data
+                    .write_u64::<BigEndian>(value)
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+            }
+            _ => {
+                return Err(DB3Error::DocumentDecodeError(
+                    "value type is not supported".to_string(),
+                ))
+            }
+        }
+        Ok(())
+    }
+
+    pub fn new() -> Self {
+        Self { data: Vec::new() }
+    }
+
+    pub fn create(fields: &Vec<Option<Bson>>) -> std::result::Result<Self, DB3Error> {
+        let mut key = Self::new();
+        if fields.len() > 16 {
+            return Err(DB3Error::DocumentDecodeError(format!(
+                "field length is over 16"
+            )));
+        }
+        for field in fields {
+            key.add(&field)?;
+        }
+        Ok(key)
+    }
+    pub fn create_single_key(field: Option<Bson>) -> std::result::Result<Self, DB3Error> {
+        Self::create(&vec![field])
+    }
+    fn read_next_field(
+        cursor: &mut Cursor<Vec<u8>>,
+    ) -> std::result::Result<Option<Bson>, DB3Error> {
+        let field_type = cursor
+            .read_u8()
+            .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+        match field_type {
+            0x01 => Ok(None),
+            0x02 => {
+                let value = cursor
+                    .read_u8()
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                Ok(Some(Bson::Boolean(value != 0)))
+            }
+            0x03 => {
+                let value = cursor
+                    .read_u32::<BigEndian>()
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                Ok(Some(Bson::Int32(Self::decode_u32(value))))
+            }
+            0x04 => {
+                let value = cursor
+                    .read_u64::<BigEndian>()
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                Ok(Some(Bson::Int64(Self::decode_u64(value))))
+            }
+            0x07 => {
+                let value = cursor
+                    .read_u64::<BigEndian>()
+                    .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                Ok(Some(Bson::DateTime(bson::DateTime::from_millis(
+                    Self::decode_u64(value),
+                ))))
+            }
+            0x0a => {
+                let mut buf = vec![];
+                while cursor.has_remaining() {
+                    let c = cursor
+                        .read_u8()
+                        .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                    if c == 0 {
+                        let value = String::from_utf8(buf)
+                            .map_err(|e| DB3Error::DocumentDecodeError(format!("{e}")))?;
+                        return Ok(Some(Bson::String(value)));
+                    }
+                    buf.push(c);
+                }
+
+                Err(DB3Error::DocumentDecodeError(
+                    "string field is not terminated by \0".to_string(),
+                ))
+            }
+            _ => Err(DB3Error::DocumentDecodeError(
+                "field type is not supported".to_string(),
+            )),
+        }
+    }
+    /// extract_fields extract fields from a key.
+    pub fn extract_fields(&self) -> std::result::Result<Vec<Option<Bson>>, DB3Error> {
+        if self.data.len() == 0 {
+            return Ok(Vec::new());
+        }
+        let mut fields = Vec::new();
+        let mut cursor = Cursor::new(self.data.clone());
+        while cursor.has_remaining() {
+            match Self::read_next_field(&mut cursor) {
+                Ok(field) => {
+                    fields.push(field);
+                }
+                Err(e) => return Err(e),
+            }
+        }
+        Ok(fields)
+    }
+    pub fn try_from_bytes(data: &[u8]) -> std::result::Result<Self, DB3Error> {
+        Ok(Self {
+            data: data.to_vec(),
+        })
+    }
+}
+
+impl AsRef<Vec<u8>> for FieldKey {
+    fn as_ref(&self) -> &Vec<u8> {
+        &self.data
+    }
+}
+impl fmt::Display for FieldKey {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        // Customize so only `x` and `y` are denoted.
+        match self.extract_fields() {
+            Ok(fields) => {
+                let key_string = fields
+                    .iter()
+                    .map(|f| match f {
+                        Some(Bson::String(s)) => format!("{}", s).to_string(),
+                        Some(Bson::Int32(i)) => format!("{}", i).to_string(),
+                        Some(Bson::Int64(i)) => format!("{}", i).to_string(),
+                        Some(Bson::Boolean(b)) => format!("{}", b).to_string(),
+                        Some(Bson::DateTime(d)) => format!("{}", d).to_string(),
+                        None => "null".to_string(),
+                        _ => "NA".to_string(),
+                    })
+                    .collect::<Vec<_>>()
+                    .join("#")
+                    .to_string();
+                write!(f, "{}", key_string)
+            }
+            Err(_) => write!(f, "NA"),
+        }
+    }
+}
 /// DocumentId := CollectionId + IndexFieldId + KeyBytes + DocumentEntryId
 #[derive(Eq, Default, PartialEq, Ord, PartialOrd, Clone, Debug)]
 pub struct IndexId {
@@ -360,9 +624,12 @@ impl IndexId {
         u32::from_be_bytes(x)
     }
 
-    pub fn get_key(&self) -> std::result::Result<&str, DB3Error> {
-        // TODO: format get key
-        todo!()
+    pub fn get_key(&self) -> std::result::Result<FieldKey, DB3Error> {
+        FieldKey::try_from_bytes(
+            self.data[TYPE_ID_LENGTH + OP_ENTRY_ID_LENGTH + INDEX_FIELD_ID_LENGTH
+                ..self.data.len() - DOCUMENT_ID_LENGTH]
+                .as_ref(),
+        )
     }
 }
 impl AsRef<Vec<u8>> for IndexId {
@@ -375,12 +642,13 @@ impl fmt::Display for IndexId {
         // Customize so only `x` and `y` are denoted.
         let collection_id = self.get_collection_id().map_err(|e| e).unwrap();
         let document_id = self.get_document_id().map_err(|e| e).unwrap();
+        let field_key = self.get_key().map_err(|e| e).unwrap();
         write!(
             f,
             "INDEX|{}|{}|{}|{}",
             collection_id,
             self.get_index_field_id(),
-            "KEY_HIDE",
+            field_key,
             document_id
         )
     }
@@ -477,7 +745,7 @@ impl TryFrom<(&DB3Address, u64)> for DbId {
 #[cfg(test)]
 mod tests {
     use super::*;
-
+    use chrono::Utc;
     #[test]
     fn it_works() {}
 
@@ -526,19 +794,16 @@ mod tests {
         let collection_id = CollectionId::create(1000, 100, 10).unwrap();
         let document_entry_id = DocumentEntryId::create(999, 99, 9).unwrap();
         let document_id = DocumentId::create(&collection_id, &document_entry_id).unwrap();
-
-        let index_id = IndexId::create(
-            &collection_id,
-            3,
-            "key_content".as_bytes().as_ref(),
-            &document_id,
-        )
-        .unwrap();
+        let field_key =
+            FieldKey::create_single_key(Some(Bson::String("key_content".to_string()))).unwrap();
+        let index_id =
+            IndexId::create(&collection_id, 3, &field_key.as_ref(), &document_id).unwrap();
         assert_eq!(collection_id, index_id.get_collection_id().unwrap());
         assert_eq!(document_id, index_id.get_document_id().unwrap());
         assert_eq!(3, index_id.get_index_field_id());
+        assert_eq!(field_key, index_id.get_key().unwrap());
         assert_eq!(
-            "INDEX|1000-100-10|3|KEY_HIDE|DOC|1000-100-10|999-99-9",
+            "INDEX|1000-100-10|3|key_content|DOC|1000-100-10|999-99-9",
             index_id.to_string()
         );
     }
@@ -566,4 +831,162 @@ mod tests {
         let (start, end) = BillId::get_block_range(1).unwrap();
         assert!(start.as_ref().cmp(end.as_ref()) == std::cmp::Ordering::Less);
     }
+
+    /// helper method to adapter unit test
+    fn bson_into_comparison_bytes(field: &Bson) -> std::result::Result<FieldKey, DB3Error> {
+        let key = FieldKey::create(&vec![Some(field.clone())])?;
+        let extract_field = key.extract_fields()?[0].clone().unwrap();
+        assert_eq!(extract_field, field.clone());
+        Ok(key)
+    }
+
+    #[test]
+    fn i64_bson_into_comparison_bytes_ut() {
+        let i64_neg_2 = bson_into_comparison_bytes(&Bson::Int64(-2)).unwrap();
+        let i64_neg_1 = bson_into_comparison_bytes(&Bson::Int64(-1)).unwrap();
+        let i64_small_value1 = -(0x7F00000000000000 as i64);
+        let i64_small_1 = bson_into_comparison_bytes(&Bson::Int64(i64_small_value1)).unwrap();
+        let i64_small_value2 = -(0x7000000000000000 as i64);
+        let i64_small_2 = bson_into_comparison_bytes(&Bson::Int64(i64_small_value2)).unwrap();
+        let i64_0 = bson_into_comparison_bytes(&Bson::Int64(0)).unwrap();
+        let i64_1 = bson_into_comparison_bytes(&Bson::Int64(1)).unwrap();
+        let i64_big_value1 =
+            bson_into_comparison_bytes(&Bson::Int64(0x7000000000000000 as i64)).unwrap();
+        let i64_big_value2 =
+            bson_into_comparison_bytes(&Bson::Int64(0x7F00000000000000 as i64)).unwrap();
+        let i64_max = bson_into_comparison_bytes(&Bson::Int64(i64::MAX)).unwrap();
+        let i64_min = bson_into_comparison_bytes(&Bson::Int64(i64::MIN)).unwrap();
+        println!("i64_min: {:?}", i64_min);
+        println!("{} i64_small_value1: {:?}", i64_small_value1, i64_small_1);
+        println!("{} i64_small_value2: {:?}", i64_small_value2, i64_small_2);
+        println!("i64_-2: {:?}", i64_neg_2);
+        println!("i64_-1: {:?}", i64_neg_1);
+        println!("i64_0: {:?}", i64_0);
+        println!("i64_1: {:?}", i64_1);
+        println!(
+            "{} i64_big_value1: {:?}",
+            0x7000000000000000 as i64, i64_big_value1
+        );
+        println!(
+            "{} i64_big_value2: {:?}",
+            0x7F00000000000000 as i64, i64_big_value2
+        );
+        println!("i64_max: {:?}", i64_max);
+
+        assert!(i64_min < i64_small_1);
+        assert!(i64_small_1 < i64_small_2);
+        assert!(i64_small_2 < i64_neg_1);
+        assert!(i64_neg_2 < i64_1);
+        assert!(i64_neg_1 < i64_0);
+        assert!(i64_0 < i64_1);
+        assert!(i64_1 < i64_big_value1);
+        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);
+        let i32_small_1 = bson_into_comparison_bytes(&Bson::Int32(i32_small_value1)).unwrap();
+        let i32_small_value2 = -(0x70000000 as i32);
+        let i32_small_2 = bson_into_comparison_bytes(&Bson::Int32(i32_small_value2)).unwrap();
+        let i32_neg_2 = bson_into_comparison_bytes(&Bson::Int32(-2)).unwrap();
+        let i32_neg_1 = bson_into_comparison_bytes(&Bson::Int32(-1)).unwrap();
+        let i32_0 = bson_into_comparison_bytes(&Bson::Int32(0)).unwrap();
+        let i32_1 = bson_into_comparison_bytes(&Bson::Int32(1)).unwrap();
+        let i32_big_value1 = bson_into_comparison_bytes(&Bson::Int32(0x70000000 as i32)).unwrap();
+        let i32_big_value2 = bson_into_comparison_bytes(&Bson::Int32(0x7F000000 as i32)).unwrap();
+        let i32_max = bson_into_comparison_bytes(&Bson::Int32(i32::MAX)).unwrap();
+        let i32_min = bson_into_comparison_bytes(&Bson::Int32(i32::MIN)).unwrap();
+
+        println!("i32_min: {:?}", i32_min);
+        println!("{} i32_small_1: {:?}", i32_small_value1, i32_small_1);
+        println!("{} i32_small_2: {:?}", i32_small_value2, i32_small_2);
+        println!("i32_-2: {:?}", i32_neg_2);
+        println!("i32_-1: {:?}", i32_neg_1);
+        println!("i32_0: {:?}", i32_0);
+        println!("i32_1: {:?}", i32_1);
+        println!("{} i32_big_value1: {:?}", 0x70000000 as i32, i32_big_value1);
+        println!("{} i32_big_value2: {:?}", 0x7F000000 as i32, i32_big_value2);
+        println!("i32_max: {:?}", i32_max);
+
+        assert!(i32_min < i32_small_1);
+        assert!(i32_small_1 < i32_small_2);
+        assert!(i32_small_2 < i32_neg_2);
+        assert!(i32_neg_2 < i32_neg_1);
+        assert!(i32_neg_1 < i32_0);
+        assert!(i32_0 < i32_1);
+        assert!(i32_1 < i32_big_value1);
+        assert!(i32_big_value1 < i32_big_value2);
+        assert!(i32_big_value2 < i32_max);
+    }
+
+    #[test]
+    fn string_bson_into_comparison_bytes_ut() {
+        let empty_str = bson_into_comparison_bytes(&Bson::String("".to_string())).unwrap();
+        let a_str = bson_into_comparison_bytes(&Bson::String("a".to_string())).unwrap();
+        let z_str = bson_into_comparison_bytes(&Bson::String("z".to_string())).unwrap();
+        let a_long_str = bson_into_comparison_bytes(&Bson::String("abcdefg".to_string())).unwrap();
+        println!("empty_str: {:?}", empty_str);
+        println!("a_str: {:?}", a_str);
+        println!("z_str: {:?}", z_str);
+        println!("a_long_str: {:?}", a_long_str);
+        assert!(empty_str < a_str);
+        assert!(a_str < z_str);
+        assert!(a_long_str < z_str);
+    }
+
+    #[test]
+    fn datetime_bson_into_comparison_bytes_ut() {
+        let now_ts = Utc::now().timestamp_millis();
+        let now = bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(now_ts)))
+            .unwrap();
+        let now_minus_one =
+            bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(now_ts - 1)))
+                .unwrap();
+        let now_plus_one =
+            bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(now_ts + 1)))
+                .unwrap();
+        let zero_ts =
+            bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::from_millis(0))).unwrap();
+        let min_ts = bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::MIN)).unwrap();
+        let max_ts = bson_into_comparison_bytes(&Bson::DateTime(bson::DateTime::MAX)).unwrap();
+
+        assert!(min_ts < zero_ts);
+        assert!(zero_ts < now_minus_one);
+        assert!(now_minus_one < now);
+        assert!(now < now_plus_one);
+        assert!(now_plus_one < max_ts);
+    }
+
+    #[test]
+    fn multiple_fields_key_ut() {
+        let fields = vec![
+            Some(Bson::String("a".to_string())),
+            Some(Bson::Int64(1 as i64)),
+        ];
+        let key = FieldKey::create(&fields).unwrap();
+        let extract_fields = key.extract_fields().unwrap();
+        assert_eq!(extract_fields[0], Some(Bson::String("a".to_string())));
+        assert_eq!(extract_fields[1], Some(Bson::Int64(1 as i64)));
+        assert_eq!("a#1", format!("{}", key));
+        let fields = vec![
+            Some(Bson::String("".to_string())),
+            Some(Bson::Int64(1 as i64)),
+        ];
+        let key = FieldKey::create(&fields).unwrap();
+        let extract_fields = key.extract_fields().unwrap();
+        assert_eq!(extract_fields[0], Some(Bson::String("".to_string())));
+        assert_eq!(extract_fields[1], Some(Bson::Int64(1 as i64)));
+        assert_eq!("#1", format!("{}", key));
+        let fields = vec![
+            Some(Bson::Int64(10 as i64)),
+            Some(Bson::String("".to_string())),
+        ];
+        let key = FieldKey::create(&fields).unwrap();
+        let extract_fields = key.extract_fields().unwrap();
+        assert_eq!(extract_fields[0], Some(Bson::Int64(10 as i64)));
+        assert_eq!(extract_fields[1], Some(Bson::String("".to_string())));
+        assert_eq!("10#", format!("{}", key));
+    }
 }
diff --git a/src/storage/src/db3_document.rs b/src/storage/src/db3_document.rs
index c9b55560..008a3db7 100644
--- a/src/storage/src/db3_document.rs
+++ b/src/storage/src/db3_document.rs
@@ -21,7 +21,7 @@ use bson::Document;
 use bson::{Binary, Bson};
 use db3_base::bson_util;
 use db3_crypto::db3_address::DB3Address;
-use db3_crypto::id::{DocumentId, TxId};
+use db3_crypto::id::{DocumentId, FieldKey, TxId};
 use db3_error::DB3Error;
 #[derive(Debug)]
 pub struct DB3Document {
@@ -135,33 +135,37 @@ impl DB3Document {
     pub fn get_keys(
         &self,
         index: &db3_proto::db3_database_proto::Index,
-    ) -> std::result::Result<Option<Vec<u8>>, DB3Error> {
+    ) -> std::result::Result<Option<FieldKey>, DB3Error> {
         let keys: Vec<_> = index.fields.iter().map(|f| f.field_path.as_str()).collect();
         match keys.len() {
             0 => Err(DB3Error::DocumentDecodeError(format!(
                 "fail to get empty keys"
             ))),
-            1 => self.get_single_key(keys[0]),
             _ => self.get_multiple_keys(keys),
         }
     }
 
-    fn get_single_key(&self, key: &str) -> std::result::Result<Option<Vec<u8>>, DB3Error> {
-        match self.get_document()?.get(key) {
-            Some(value) => Ok(Some(bson_util::bson_into_comparison_bytes(value)?)),
-            None => Ok(None),
-        }
-    }
-
-    fn get_multiple_keys(&self, keys: Vec<&str>) -> std::result::Result<Option<Vec<u8>>, DB3Error> {
-        let mut data: Vec<u8> = Vec::new();
+    fn get_multiple_keys(
+        &self,
+        keys: Vec<&str>,
+    ) -> std::result::Result<Option<FieldKey>, DB3Error> {
+        let mut fields = vec![];
+        let mut has_field = false;
         for key in keys.iter() {
-            match self.get_single_key(key)? {
-                Some(v) => data.extend_from_slice(v.as_ref()),
-                None => data.extend_from_slice("_".as_bytes()),
+            match self.get_document()?.get(key) {
+                Some(value) => {
+                    fields.push(Some(value.clone()));
+                    has_field = true;
+                }
+                None => fields.push(None),
             }
         }
-        Ok(Some(data))
+        if has_field {
+            Ok(Some(FieldKey::create(&fields)?))
+        } else {
+            // if no fields exist, return None
+            Ok(None)
+        }
     }
 }
 impl AsRef<Document> for DB3Document {
@@ -323,7 +327,8 @@ mod tests {
             fields: vec![index_field],
         };
         if let Ok(Some(keys)) = document.get_keys(&index) {
-            assert_eq!("John Doe", std::str::from_utf8(keys.as_ref()).unwrap())
+            let fields = keys.extract_fields().unwrap();
+            assert_eq!(Some(Bson::String("John Doe".to_string())), fields[0]);
         } else {
             assert!(false);
         }
@@ -355,7 +360,10 @@ mod tests {
             fields: vec![index_field],
         };
         if let Ok(Some(keys)) = document.get_keys(&index) {
-            assert_eq!("John Doe", std::str::from_utf8(keys.as_ref()).unwrap())
+            assert_eq!(
+                Some(Bson::String("John Doe".to_string())),
+                keys.extract_fields().unwrap()[0]
+            );
         } else {
             assert!(false);
         }
@@ -387,7 +395,7 @@ mod tests {
             fields: vec![index_field],
         };
         if let Ok(Some(keys)) = document.get_keys(&index) {
-            assert_eq!(vec![128, 0, 0, 0, 0, 0, 0, 43], keys);
+            assert_eq!(vec![4, 128, 0, 0, 0, 0, 0, 0, 43], *keys.as_ref());
         } else {
             assert!(false);
         }
@@ -456,12 +464,11 @@ mod tests {
             ],
         };
         if let Ok(Some(keys)) = document.get_keys(&index) {
-            let str_len = "John Doe".to_string().as_str().len();
             assert_eq!(
-                "John Doe",
-                std::str::from_utf8(keys[0..str_len].as_ref()).unwrap()
+                Some(Bson::String("John Doe".to_string())),
+                keys.extract_fields().unwrap()[0]
             );
-            assert_eq!([128, 0, 0, 0, 0, 0, 0, 43], keys[str_len..]);
+            assert_eq!(Some(Bson::Int64(43)), keys.extract_fields().unwrap()[1]);
         } else {
             assert!(false);
         }
diff --git a/src/storage/src/db_store.rs b/src/storage/src/db_store.rs
index fbdcabf7..c2e80ce9 100644
--- a/src/storage/src/db_store.rs
+++ b/src/storage/src/db_store.rs
@@ -21,7 +21,7 @@ use bytes::BytesMut;
 use db3_base::bson_util;
 use db3_crypto::{
     db3_address::DB3Address, id::CollectionId, id::DbId, id::DocumentEntryId, id::DocumentId,
-    id::IndexId, id::TxId,
+    id::FieldKey, id::IndexId, id::TxId,
 };
 use db3_error::{DB3Error, Result};
 use db3_proto::db3_database_proto::index::IndexField;
@@ -605,7 +605,7 @@ impl DbStore {
                                     let index_id = IndexId::create(
                                         &collection_id,
                                         index.id,
-                                        &old_key,
+                                        &old_key.as_ref(),
                                         &document_id,
                                     )?;
                                     // remove old index entry
@@ -620,7 +620,7 @@ impl DbStore {
                                     let index_id = IndexId::create(
                                         &collection_id,
                                         index.id,
-                                        &new_key,
+                                        &new_key.as_ref(),
                                         &document_id,
                                     )?;
                                     // add new index entry
@@ -741,9 +741,9 @@ impl DbStore {
                                 };
 
                                 let key = match &field_filter.value {
-                                    Some(value) => bson_util::bson_into_comparison_bytes(
-                                        &bson_util::bson_value_from_proto_value(value)?,
-                                    )?,
+                                    Some(value) => FieldKey::create_single_key(Some(
+                                        bson_util::bson_value_from_proto_value(value)?,
+                                    ))?,
                                     None => {
                                         return Err(DB3Error::InvalidFilterValue(
                                             "None field filter value un-support".to_string(),
@@ -753,7 +753,7 @@ impl DbStore {
                                 let range = Self::generate_range_with_single_field_filter(
                                     &collection_id,
                                     index,
-                                    &key,
+                                    key.as_ref(),
                                     Operator::from_i32(field_filter.op),
                                 )?;
                                 Self::execute_query(db, &range, limit)
@@ -761,8 +761,8 @@ impl DbStore {
                             Some(FilterType::CompositeFilter(composite_filter)) => {
                                 match CompositeOp::from_i32(composite_filter.op) {
                                     Some(CompositeOp::And) => {
-                                        let mut filter_names = vec![];
-                                        let mut keys = vec![];
+                                        let mut fields_names = vec![];
+                                        let mut fields_values = vec![];
 
                                         for filter in composite_filter.filters.iter() {
                                             match &filter.filter_type {
@@ -776,12 +776,11 @@ impl DbStore {
                                                                     "CompositeOp And only support Equal".to_string(),
                                                                 ));
                                                             }
-                                                            filter_names.push(
+                                                            fields_names.push(
                                                                 field_filter.field.to_string(),
                                                             );
-                                                            keys.push(bson_util::bson_into_comparison_bytes(
-                                                                &bson_util::bson_value_from_proto_value(value)?,
-                                                            )?);
+                                                            fields_values.push(Some(
+                                                                bson_util::bson_value_from_proto_value(value)?));
                                                         }
                                                         None => {
                                                             return Err(DB3Error::InvalidFilterValue(
@@ -799,18 +798,18 @@ impl DbStore {
                                             }
                                         }
 
-                                        if filter_names.is_empty() {
+                                        if fields_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]) {
+                                        let key = FieldKey::create(&fields_values)?;
+                                        let index = match field_index_map.get(&fields_names[0]) {
                                             Some(index_list) => {
                                                 match index_list.iter().find_or_first(|i| {
                                                     Self::index_start_with_fields(
-                                                        &filter_names,
+                                                        &fields_names,
                                                         &i.fields,
                                                     )
                                                 }) {
@@ -818,7 +817,7 @@ impl DbStore {
                                                     None => {
                                                         return Err(
                                                             DB3Error::IndexNotFoundForFiledFilter(
-                                                                filter_names[0].to_string(),
+                                                                fields_names[0].to_string(),
                                                             ),
                                                         );
                                                     }
@@ -826,23 +825,17 @@ impl DbStore {
                                             }
                                             None => {
                                                 return Err(DB3Error::IndexNotFoundForFiledFilter(
-                                                    filter_names[0].to_string(),
+                                                    fields_names[0].to_string(),
                                                 ));
                                             }
                                         };
-                                        debug!("filter names: {:?}", filter_names);
-                                        debug!("keys: {:?}", keys);
+                                        debug!("filter names: {:?}", fields_names);
                                         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,
+                                            key.as_ref(),
                                             Some(Operator::Equal),
                                         )?;
                                         debug!("range: {:?}", range);
@@ -1563,6 +1556,16 @@ mod tests {
                 "+44 1234567",
                 "+44 2345678"
             ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "",
+            "age": 45,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
         }"#
                 .to_string(),
             ],
@@ -1572,7 +1575,7 @@ mod tests {
         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());
+        assert_eq!(5, ids.len());
 
         // test query with composite filter
         for (name, age, exp) in [
@@ -1584,6 +1587,10 @@ mod tests {
             ("Bill", 46, vec![]),
             // Select * from collection where name = "Mike" and age = 44
             ("Mike", 44, vec![("Mike", 44)]),
+            // Select * from collection where name = "" and age = 45
+            ("", 45, vec![("", 45)]),
+            // Select * from collection where name = "" and age = 46
+            ("", 46, vec![]),
         ] {
             let query = StructuredQuery {
                 collection_name: collection_name.to_string(),
@@ -1698,6 +1705,16 @@ mod tests {
                 "+44 1234567",
                 "+44 2345678"
             ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "",
+            "age": 45,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
         }"#
                 .to_string(),
             ],
@@ -1707,7 +1724,7 @@ mod tests {
         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());
+        assert_eq!(5, ids.len());
 
         // test query with composite filter
         for (name, name_op, age, age_op, composite_op, is_ok) in [
@@ -1720,6 +1737,14 @@ mod tests {
                 CompositeOp::And,
                 true,
             ),
+            (
+                "",
+                Operator::Equal,
+                45,
+                Operator::Equal,
+                CompositeOp::And,
+                true,
+            ),
             (
                 "Bill",
                 Operator::Equal,
@@ -1841,6 +1866,16 @@ mod tests {
                 "+44 1234567",
                 "+44 2345678"
             ]
+        }"#
+                .to_string(),
+                r#"
+        {
+            "name": "",
+            "age": 45,
+            "phones": [
+                "+44 1234567",
+                "+44 2345678"
+            ]
         }"#
                 .to_string(),
             ],
@@ -1850,7 +1885,7 @@ mod tests {
         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());
+        assert_eq!(5, ids.len());
 
         // run query db not exist
         let query = StructuredQuery {
@@ -1882,7 +1917,7 @@ mod tests {
         };
         let res = DbStore::run_query(db.as_ref(), &db_id, &query);
         assert!(res.is_ok(), "{:?}", res);
-        assert_eq!(4, res.unwrap().len());
+        assert_eq!(5, res.unwrap().len());
 
         // run query: select * from collection limit 2
         let query = StructuredQuery {
@@ -1896,13 +1931,19 @@ mod tests {
 
         // test query with ==, <, <=, >, >= condition
         for (field, value_str, op, exp) in [
+            ("name", "", Operator::Equal, vec![""]),
             ("name", "Bill", Operator::Equal, vec!["Bill", "Bill"]),
-            ("name", "John Doe", Operator::LessThan, vec!["Bill", "Bill"]),
+            (
+                "name",
+                "John Doe",
+                Operator::LessThan,
+                vec!["", "Bill", "Bill"],
+            ),
             (
                 "name",
                 "John Doe",
                 Operator::LessThanOrEqual,
-                vec!["Bill", "Bill", "John Doe"],
+                vec!["", "Bill", "Bill", "John Doe"],
             ),
             (
                 "name",
@@ -2616,12 +2657,12 @@ mod tests {
             name: "idx1".to_string(),
             fields: vec![index_field_name],
         };
-        let key = bson_util::bson_into_comparison_bytes(&Bson::String("Bill".to_string())).unwrap();
+        let key = FieldKey::create_single_key(Some(Bson::String("Bill".to_string()))).unwrap();
 
         let res = DbStore::generate_range_with_single_field_filter(
             &collection_id,
             &index_name,
-            &key,
+            &key.as_ref(),
             Some(Operator::Equal),
         );
         assert!(res.is_ok(), "{:?}", res);
@@ -2647,7 +2688,7 @@ mod tests {
         let res = DbStore::generate_range_with_single_field_filter(
             &collection_id,
             &index_name,
-            &key,
+            &key.as_ref(),
             Some(Operator::GreaterThan),
         );
         assert!(res.is_ok(), "{:?}", res);
@@ -2674,7 +2715,7 @@ mod tests {
         let res = DbStore::generate_range_with_single_field_filter(
             &collection_id,
             &index_name,
-            &key,
+            &key.as_ref(),
             Some(Operator::GreaterThanOrEqual),
         );
         assert!(res.is_ok(), "{:?}", res);
@@ -2701,7 +2742,7 @@ mod tests {
         let res = DbStore::generate_range_with_single_field_filter(
             &collection_id,
             &index_name,
-            &key,
+            &key.as_ref(),
             Some(Operator::LessThan),
         );
         assert!(res.is_ok(), "{:?}", res);
@@ -2728,7 +2769,7 @@ mod tests {
         let res = DbStore::generate_range_with_single_field_filter(
             &collection_id,
             &index_name,
-            &key,
+            &key.as_ref(),
             Some(Operator::LessThanOrEqual),
         );
         assert!(res.is_ok(), "{:?}", res);


Test patch: