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

Issue:
Standalone-indexer: adapt to db3 light node mutation api 
## 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/node/src/indexer_impl.rs b/src/node/src/indexer_impl.rs
index e6e33d4c..637b412c 100644
--- a/src/node/src/indexer_impl.rs
+++ b/src/node/src/indexer_impl.rs
@@ -20,7 +20,9 @@ use db3_base::bson_util::bytes_to_bson_document;
 use db3_crypto::db3_address::DB3Address;
 use db3_error::{DB3Error, Result};
 use db3_proto::db3_indexer_proto::indexer_node_server::IndexerNode;
-use db3_proto::db3_indexer_proto::{IndexerStatus, ShowIndexerStatusRequest};
+use db3_proto::db3_indexer_proto::{
+    IndexerStatus, RunQueryRequest, RunQueryResponse, ShowIndexerStatusRequest,
+};
 use db3_proto::db3_mutation_v2_proto::mutation::body_wrapper::Body;
 use db3_proto::db3_mutation_v2_proto::MutationAction;
 use db3_proto::db3_storage_proto::block_response::MutationWrapper;
@@ -73,7 +75,7 @@ impl IndexerNodeImpl {
         match event.event {
             Some(event_message::Event::BlockEvent(be)) => {
                 info!(
-                    "[IndexerBlockSyncer] Receive BlockEvent: Block\t{}MutationCount\t{}",
+                    "Receive BlockEvent: Block\t{}\tMutationCount\t{}",
                     be.block_id, be.mutation_count,
                 );
                 let response = store_sdk
@@ -265,6 +267,34 @@ impl IndexerNode for IndexerNodeImpl {
     ) -> std::result::Result<Response<IndexerStatus>, Status> {
         Err(Status::internal("err".to_string()))
     }
+
+    async fn run_query(
+        &self,
+        request: Request<RunQueryRequest>,
+    ) -> std::result::Result<Response<RunQueryResponse>, Status> {
+        let r = request.into_inner();
+        let addr =
+            DB3Address::from_hex(r.db.as_str()).map_err(|e| Status::internal(format!("{e}")))?;
+        if let Some(q) = &r.query {
+            info!("query str {} q {:?}", q.query_str, q);
+            let documents = self
+                .db_store
+                .query_docs(&addr, r.col_name.as_str(), q)
+                .map_err(|e| Status::internal(format!("{e}")))?;
+
+            info!(
+                "query str {} from collection {} in db {} with result len {}, parameters len {}",
+                q.query_str,
+                r.col_name.as_str(),
+                r.db.as_str(),
+                documents.len(),
+                q.parameters.len()
+            );
+            Ok(Response::new(RunQueryResponse { documents }))
+        } else {
+            Err(Status::internal("no query provided".to_string()))
+        }
+    }
 }
 #[cfg(test)]
 mod tests {}
diff --git a/src/proto/proto/db3_database_v2.proto b/src/proto/proto/db3_database_v2.proto
index d7326efb..ad89574f 100644
--- a/src/proto/proto/db3_database_v2.proto
+++ b/src/proto/proto/db3_database_v2.proto
@@ -71,10 +71,9 @@ message Index {
 }
 
 message Document {
-  bytes id = 1;
-  bytes doc = 2;
-  bytes owner = 3;
-  bytes tx_id = 4;
+  int64 id = 1;
+  string doc = 2;
+  string owner = 3;
 }
 
 message QueryParameter {
@@ -84,6 +83,7 @@ message QueryParameter {
     bool bool_value = 3;
     string str_value = 4;
   }
+  int32 idx = 5;
 }
 
 message Query {
@@ -91,6 +91,7 @@ message Query {
   repeated QueryParameter parameters = 2;
 }
 
+
 // A Firestore query.
 message StructuredQuery {
   // A filter.
diff --git a/src/proto/proto/db3_indexer.proto b/src/proto/proto/db3_indexer.proto
index 6cdaea7b..f6e42651 100644
--- a/src/proto/proto/db3_indexer.proto
+++ b/src/proto/proto/db3_indexer.proto
@@ -18,7 +18,7 @@
 syntax = "proto3";
 package db3_indexer_proto;
 
-import "db3_database.proto";
+import "db3_database_v2.proto";
 
 message IndexerStatus {
   uint64 total_database_count = 1;
@@ -29,36 +29,13 @@ message IndexerStatus {
   uint64 total_storage_in_bytes = 7;
 }
 
-message ShowDatabaseRequest {
-  // the database address in hex format
-  string address = 2;
-  // the owner address in hex format
-  string owner_address = 3;
-}
-
-message ShowDatabaseResponse {
-  repeated db3_database_proto.Database dbs = 1;
-}
-
 message RunQueryRequest {
-  // a hex string
-  string address = 2;
-  db3_database_proto.StructuredQuery query = 3;
-  // TODO: support paging setting (e.g page_size, page_token)
-  // TODO: support document mask
+  string db = 1;
+  string col_name = 2;
+  db3_database_v2_proto.Query query = 3;
 }
-message GetDocumentRequest {
-  // a hex string
-  string id = 2;
-}
-
-message GetDocumentResponse {
-  db3_database_proto.Document document = 1;
-}
-
 message RunQueryResponse {
-  repeated db3_database_proto.Document documents = 2;
-  // TODO: support next_page_token for page loading
+  repeated db3_database_v2_proto.Document documents = 2;
 }
 
 message ShowIndexerStatusRequest {}
@@ -67,9 +44,9 @@ service IndexerNode {
   // method for show indexer status
   rpc ShowIndexerStatus(ShowIndexerStatusRequest) returns (IndexerStatus) {}
   // method for show database
-//  rpc ShowDatabase(ShowDatabaseRequest) returns (ShowDatabaseResponse) {}
-//  // method for query document
-//  rpc RunQuery(RunQueryRequest) returns (RunQueryResponse) {}
-//  // method for get document
-//  rpc GetDocument(GetDocumentRequest) returns (GetDocumentResponse) {}
+  //  rpc ShowDatabase(ShowDatabaseRequest) returns (ShowDatabaseResponse) {}
+  //  // method for query document
+  rpc RunQuery(RunQueryRequest) returns (RunQueryResponse) {}
+  //  // method for get document
+  //  rpc GetDocument(GetDocumentRequest) returns (GetDocumentResponse) {}
 }
diff --git a/src/sdk/src/indexer_sdk.rs b/src/sdk/src/indexer_sdk.rs
index 565c44c0..3aafeb63 100644
--- a/src/sdk/src/indexer_sdk.rs
+++ b/src/sdk/src/indexer_sdk.rs
@@ -15,14 +15,14 @@
 // limitations under the License.
 //
 
-use db3_proto::db3_database_proto::structured_query::{Limit, Projection};
-use db3_proto::db3_database_proto::{Database, Document, StructuredQuery};
-use db3_proto::db3_indexer_proto::{
-    indexer_node_client::IndexerNodeClient, GetDocumentRequest, IndexerStatus, RunQueryRequest,
-    RunQueryResponse, ShowDatabaseRequest, ShowIndexerStatusRequest,
-};
-use std::sync::Arc;
-use tonic::Status;
+//use db3_proto::db3_database_proto::structured_query::{Limit, Projection};
+//use db3_proto::db3_database_proto::{Database, Document, StructuredQuery};
+//use db3_proto::db3_indexer_proto::{
+//    indexer_node_client::IndexerNodeClient, GetDocumentRequest, IndexerStatus, RunQueryRequest,
+//    RunQueryResponse, ShowDatabaseRequest, ShowIndexerStatusRequest,
+//};
+//use std::sync::Arc;
+//use tonic::Status;
 //
 // pub struct IndexerSDK {
 //     client: Arc<IndexerNodeClient<tonic::transport::Channel>>,
diff --git a/src/storage/src/db_store_v2.rs b/src/storage/src/db_store_v2.rs
index b2ab9108..318ea3d1 100644
--- a/src/storage/src/db_store_v2.rs
+++ b/src/storage/src/db_store_v2.rs
@@ -24,7 +24,7 @@ use crate::collection_key;
 use crate::doc_store::{DocStore, DocStoreConfig};
 use db3_error::{DB3Error, Result};
 use db3_proto::db3_database_v2_proto::{
-    database_message, Collection, DatabaseMessage, DocumentDatabase,
+    database_message, Collection, DatabaseMessage, Document, DocumentDatabase, Query,
 };
 use db3_proto::db3_mutation_v2_proto::{CollectionMutation, DocumentDatabaseMutation};
 use prost::Message;
@@ -262,7 +262,7 @@ impl DBStoreV2 {
     pub fn update_docs(
         &self,
         db_addr: &DB3Address,
-        sender: &DB3Address,
+        _sender: &DB3Address,
         col_name: &str,
         docs: &[(String, i64)],
     ) -> Result<()> {
@@ -290,10 +290,50 @@ impl DBStoreV2 {
             Ok(())
         }
     }
+
+    pub fn query_docs(
+        &self,
+        db_addr: &DB3Address,
+        col_name: &str,
+        query: &Query,
+    ) -> Result<Vec<Document>> {
+        let ck = collection_key::build_collection_key(db_addr, col_name)
+            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
+        let collection_store_cf_handle = self
+            .se
+            .cf_handle(self.config.collection_store_cf_name.as_str())
+            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
+        let ck_ref: &[u8] = ck.as_ref();
+        let value = self
+            .se
+            .get_cf(&collection_store_cf_handle, ck_ref)
+            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
+        if let None = value {
+            return Err(DB3Error::ReadStoreError(format!(
+                "collection with name {} does not exist",
+                col_name
+            )));
+        }
+        if self.config.enable_doc_store {
+            let docs = self.doc_store.execute_query(db_addr, col_name, query)?;
+            //TODO add owner
+            Ok(docs
+                .iter()
+                .map(|doc| Document {
+                    id: doc.0,
+                    doc: doc.1.to_string(),
+                    owner: "".to_string(),
+                })
+                .collect())
+        } else {
+            Ok(vec![])
+        }
+    }
+
     pub fn delete_docs(
         &self,
         db_addr: &DB3Address,
-        sender: &DB3Address,
+        _sender: &DB3Address,
         col_name: &str,
         ids: &[i64],
     ) -> Result<()> {
@@ -325,7 +365,7 @@ impl DBStoreV2 {
     pub fn add_docs(
         &self,
         db_addr: &DB3Address,
-        sender: &DB3Address,
+        _sender: &DB3Address,
         col_name: &str,
         docs: &Vec<String>,
     ) -> Result<Vec<i64>> {
diff --git a/src/storage/src/doc_store.rs b/src/storage/src/doc_store.rs
index 5e76c132..22b77bb5 100644
--- a/src/storage/src/doc_store.rs
+++ b/src/storage/src/doc_store.rs
@@ -179,21 +179,21 @@ impl DocStore {
         prepared_statement
             .init()
             .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
-        for (i, param) in query.parameters.iter().enumerate() {
+        for param in query.parameters.iter() {
             match &param.parameter {
                 Some(query_parameter::Parameter::Int64Value(v)) => {
                     prepared_statement
-                        .set_placeholder(param.name.as_str(), i as i32, *v)
+                        .set_placeholder(param.name.as_str(), param.idx, *v)
                         .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
                 }
                 Some(query_parameter::Parameter::BoolValue(v)) => {
                     prepared_statement
-                        .set_placeholder(param.name.as_str(), i as i32, *v)
+                        .set_placeholder(param.name.as_str(), param.idx, *v)
                         .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
                 }
                 Some(query_parameter::Parameter::StrValue(v)) => {
                     prepared_statement
-                        .set_placeholder(param.name.as_str(), i as i32, v.as_str())
+                        .set_placeholder(param.name.as_str(), param.idx, v.as_str())
                         .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
                 }
                 _ => {}
@@ -389,6 +389,7 @@ mod tests {
                 parameters: vec![QueryParameter {
                     name: "f1".to_string(),
                     parameter: Some(query_parameter::Parameter::StrValue("f1".to_string())),
+                    idx: 0,
                 }],
             };
             if let Ok(result) = doc_store.execute_query(&DB3Address::ZERO, "col1", &query) {
@@ -400,6 +401,7 @@ mod tests {
                 parameters: vec![QueryParameter {
                     name: "f1".to_string(),
                     parameter: Some(query_parameter::Parameter::StrValue("f2".to_string())),
+                    idx: 0,
                 }],
             };
             if let Ok(result) = doc_store.execute_query(&DB3Address::ZERO, "col1", &query) {
@@ -411,6 +413,7 @@ mod tests {
                 parameters: vec![QueryParameter {
                     name: "f1".to_string(),
                     parameter: Some(query_parameter::Parameter::StrValue("f1".to_string())),
+                    idx: 0,
                 }],
             };
             if let Ok(result) = doc_store.execute_query(&DB3Address::ZERO, "col1", &query) {
diff --git a/tools/start_localnet.sh b/tools/start_localnet.sh
index 448aaa2e..ca4b0d35 100644
--- a/tools/start_localnet.sh
+++ b/tools/start_localnet.sh
@@ -66,7 +66,7 @@ echo "start tendermint node..."
 sleep 1
 
 echo "start db3 store..."
-../target/${BUILD_MODE}/db3 store --rollup-interval 60000 >store.log 2>&1  &
+../target/${BUILD_MODE}/db3 store --rollup-interval 60000 --block-interval=500 >store.log 2>&1  &
 sleep 1
 
 echo "start db3 indexer..."


Test patch: