[feat][serial] Optimize serial.

Author: nokiaMS

src/coprocessor/coprocessor.cc

@@ -312,7 +312,8 @@ butil::Status Coprocessor::DoExecute(const pb::common::KeyValue& kv, bool* has_r
   int ret = 0;
   try {
     // decode some column. not decode all
-    ret = original_record_decoder.Decode(kv.key(), kv.value(), selection_column_indexes_, original_record);
+    ret = original_record_decoder.Decode(kv.key(), kv.value(), selection_column_indexes_,
+                                         selection_column_indexes_serial_, original_record);
   } catch (const std::exception& my_exception) {
     std::string error_message = fmt::format("serial::Decode failed exception : {}", my_exception.what());
     DINGO_LOG(ERROR) << error_message;
@@ -657,6 +658,7 @@ void Coprocessor::Close() {
 
   original_column_indexes_.clear();
   selection_column_indexes_.clear();
+  selection_column_indexes_serial_.clear();
 
   if (original_serial_schemas_sorted_) {
     original_serial_schemas_sorted_.reset();
@@ -804,6 +806,7 @@ void Coprocessor::GetSelectionColumnIndexes() {
       int i = index;
       DINGO_LOG(DEBUG) << "i:" << i;
       selection_column_indexes_.push_back(i);
+      selection_column_indexes_serial_[original_column_indexes_[i]] = i;
     }
     // sort and unique
     // std::sort(selection_column_indexes_.begin(), selection_column_indexes_.end(), [](int i, int j) { return i < j;
@@ -818,6 +821,7 @@ void Coprocessor::GetSelectionColumnIndexes() {
       int i = index;
       DINGO_LOG(DEBUG) << "index:" << i;
       selection_column_indexes_.push_back(original_column_indexes_[i]);
+      selection_column_indexes_serial_[original_column_indexes_[i]] = i;
     }
     // sort and unique
     // std::sort(selection_column_indexes_.begin(), selection_column_indexes_.end(), [](int i, int j) { return i < j;

src/coprocessor/coprocessor.h

@@ -95,6 +95,7 @@ class Coprocessor : public RawCoprocessor {
   std::shared_ptr<AggregationIterator> aggregation_iterator_;
   std::vector<int> original_column_indexes_;
   std::vector<int> selection_column_indexes_;
+  std::unordered_map<int, int> selection_column_indexes_serial_;
 
   std::shared_ptr<std::vector<std::shared_ptr<BaseSchema>>> original_serial_schemas_sorted_;
   std::shared_ptr<std::vector<std::shared_ptr<BaseSchema>>> selection_serial_schemas_sorted_;

src/coprocessor/coprocessor_v2.cc

@@ -413,6 +413,7 @@ void CoprocessorV2::Close() {
   original_serial_schemas_.reset();
   original_column_indexes_.clear();
   selection_column_indexes_.clear();
+  selection_column_indexes_serial_.clear();
   result_serial_schemas_.reset();
   result_record_encoder_.reset();
   original_record_decoder_.reset();
@@ -575,7 +576,8 @@ butil::Status CoprocessorV2::DoRelExprCoreWrapper(const std::string& key, const
     int ret = 0;
     try {
       // decode some column. not decode all
-      ret = original_record_decoder_->Decode(key, value, selection_column_indexes_, original_record);
+      ret = original_record_decoder_->Decode(key, value, selection_column_indexes_,
+                                             selection_column_indexes_serial_, original_record);
     } catch (const std::exception& my_exception) {
       std::string error_message = fmt::format("serial::Decode failed exception : {}", my_exception.what());
       DINGO_LOG(ERROR) << error_message;
@@ -722,6 +724,7 @@ void CoprocessorV2::GetSelectionColumnIndexes() {
       int i = index;
       DINGO_LOG(DEBUG) << "index:" << i;
       selection_column_indexes_.push_back(original_column_indexes_[i]);
+      selection_column_indexes_serial_[original_column_indexes_[i]] = i;
     }
   } else {
     DINGO_LOG(DEBUG) << "selection_columns empty()";

src/coprocessor/coprocessor_v2.h

@@ -22,6 +22,7 @@
 #include <memory>
 #include <string>
 #include <vector>
+#include <unordered_map>
 
 #include "butil/status.h"
 #include "coprocessor/raw_coprocessor.h"
@@ -106,6 +107,7 @@ class CoprocessorV2 : public RawCoprocessor {
   std::vector<int> original_column_indexes_;  // NOLINT
   // index = dummy ; value =  original schema index
   std::vector<int> selection_column_indexes_;                                        // NOLINT
+  std::unordered_map<int, int> selection_column_indexes_serial_;                                 // NOLINT
   std::shared_ptr<std::vector<std::shared_ptr<BaseSchema>>> result_serial_schemas_;  // NOLINT
   std::shared_ptr<RecordEncoder> result_record_encoder_;                             // NOLINT
   std::shared_ptr<RecordDecoder> original_record_decoder_;                           // NOLINT