save result data to data/result directory for test case (#512)

data/client_e.csv

-company,id,executed amount,execution court,date,case number
-A big company 1,91340100MA2NFYKT09,1万元,xxx局,2020/11/30,(xx)应急罚﹝2020﹞15号
-A big company 2,,,,,
-A big company 3,91530100216582396T,128318,xxxx人民法院,2022/4/15,（2022）云xx执xx号
-A big company 4,,,,,
-A big company 5,,,,,
-A big company 6,,,,,
-A big company 7,,,,,
-A big company 8 ,11370685MB28239162,罚款人民币肆万元整（￥40000.00）。,xx市综合行政执法局,2022/3/18,招综执罚﹝2022﹞6000x号
-A big company 9 ,,,,,
-A big company 10,,,,,
+id,mean radius,mean texture
+X0,10.05,17.5
+X1,10.05,17.5
+X2,10.05,17.5
+X3,10.05,17.5
+X4,10.05,17.5
+X5,10.05,17.5
+X6,10.05,17.5
+X7,10.05,17.5
+X8,10.05,17.5
+X9,10.05,17.5
+X10,10.05,17.5
+X11,10.05,17.5
+X12,10.05,17.5
+X13,10.05,17.5
+X14,10.05,17.5
+X15,10.05,17.5
+X16,10.05,17.5
+X17,10.05,17.5
+X18,10.05,17.5
+X19,10.05,17.5
+X20,10.05,17.5

data/server_e.csv

-guaranteetype,warrantee
-银行保函,A big company
-银行保函,A big company 1
-银行保函,A big company 2
-银行保函,A big company 3
-银行保函,A big company 4
-银行保函,A big company 5
-银行保函,A big company 6
-银行保函,A big company 7
-银行保函,A big company 8
-银行保函,A big company 9
-银行保函,A big company 10
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-公司保函,A big company
-公司保函,A big company
-公司保函,A big company
-公司保函,A big company
-公司保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
-银行保函,A big company
+id,mean radius,mean texture
+X0,10.05,17.5
+X1,10.05,17.5
+X2,10.05,17.5
+X3,10.05,17.5
+X4,10.05,17.5
+X5,10.05,17.5
+X6,10.05,17.5
+X7,10.05,17.5
+X8,10.05,17.5
+X9,10.05,17.5
+X10,10.05,17.5
+X11,10.05,17.5
+X12,10.05,17.5
+X13,10.05,17.5
+X14,10.05,17.5
+X15,10.05,17.5
+X16,10.05,17.5
+X17,10.05,17.5
+X18,10.05,17.5
+X19,10.05,17.5
+X20,10.05,17.5
+X21,10.05,17.5
+X22,10.05,17.5
+X23,10.05,17.5
+X24,10.05,17.5
+X25,10.05,17.5
+X26,10.05,17.5
+X27,10.05,17.5
+X28,10.05,17.5
+X29,10.05,17.5
+X30,10.05,17.5
+X31,10.05,17.5
+X32,10.05,17.5
+X33,10.05,17.5
+X34,10.05,17.5
+X35,10.05,17.5
+X36,10.05,17.5
+X37,10.05,17.5
+X38,10.05,17.5
+X39,10.05,17.5
+X40,10.05,17.5
+X41,10.05,17.5
+X42,10.05,17.5
+X43,10.05,17.5
+X44,10.05,17.5
+X45,10.05,17.5
+X46,10.05,17.5
+X47,10.05,17.5
+X48,10.05,17.5
+X49,10.05,17.5
+X50,10.05,17.5

example/keyword_pir_task_conf.json

@@ -8,6 +8,7 @@
   },
   "params": {
     "clientData": {
+      "description": "query data",
       "type": "STRING",
       "value": [
         "SXUiwPmLNohCROXPMZIqbnLrfhCtREPzCiDYZaDghlkfGBCTqyPdqnjoaWzyNzBT",
@@ -21,6 +22,7 @@
       "value": 1
     },
     "outputFullFilename": {
+      "description": "path for client save query result",
       "type": "STRING",
       "value": "data/result/pir_result.csv"
     }
@@ -29,6 +31,5 @@
     "SERVER": {
       "SERVER": "keyword_pir_server_data"
     }
-  },
-  "input_datasets": []
+  }
 }

example/mpc_add_task_conf.json

@@ -9,7 +9,7 @@
     "params": {
         "ResFileName": {
             "type": "STRING",
-            "value": "mpc_add_result"
+            "value": "data/result/mpc_add_result.csv"
         },
         "Expr": {
             "type": "STRING",

example/mpc_div_task_conf.json

@@ -9,7 +9,7 @@
     "params": {
         "ResFileName": {
             "type": "STRING",
-            "value": "mpc_div_result"
+            "value": "data/result/mpc_div_result.csv"
         },
         "Expr": {
             "type": "STRING",

example/mpc_lr_task_conf.json

@@ -15,25 +15,9 @@
       "type": "INT32",
       "value": 100
     },
-    "modelFileName": {
+    "modelName": {
       "type": "STRING",
-      "value": "./model"
-    },
-    "hostLookupTable": {
-      "type": "STRING",
-      "value": "./hostlookuptable.csv"
-    },
-    "guestLookupTable": {
-      "type": "STRING",
-      "value": "./guestlookuptable.csv"
-    },
-    "predictFileName": {
-      "type": "STRING",
-      "value": "./prediction.csv"
-    },
-    "indicatorFileName": {
-      "type": "STRING",
-      "value": "./indicator.csv"
+      "value": "data/result/mpc_lr_model.csv"
     }
   },
   "party_datasets": {
@@ -49,6 +33,5 @@
       "Data_File": "train_party_2",
       "TestData" : "test_party_2"
     }
-  },
-  "input_datasets": []
+  }
 }
\ No newline at end of file

example/mpc_missing_value_task_conf.json

@@ -16,7 +16,8 @@
             "x2": 2,
             "x3": 2
           },
-          "newDataSetId": "train_party_missing_0"
+          "newDataSetId": "train_party_missing_0",
+          "newDataPath": "data/result/train_party_missing_0.csv"
         },
         "train_party_1": {
           "columns": {
@@ -24,7 +25,8 @@
             "x2": 2,
             "x3": 2
           },
-          "newDataSetId": "train_party_missing_1"
+          "newDataSetId": "train_party_missing_1",
+          "newDataPath": "data/result/train_party_missing_1.csv"
         },
         "train_party_2": {
           "columns": {
@@ -32,7 +34,8 @@
             "x2": 2,
             "x3": 2
           },
-          "newDataSetId": "train_party_missing_2"
+          "newDataSetId": "train_party_missing_2",
+          "newDataPath": "data/result/train_party_missing_2.csv"
         }
       }
     }

example/mpc_mul_task_conf.json

@@ -9,7 +9,7 @@
     "params": {
         "ResFileName": {
             "type": "STRING",
-            "value": "mpc_mul_result"
+            "value": "data/result/mpc_mul_result.csv"
         },
         "Expr": {
             "type": "STRING",

example/mpc_sub_task_conf.json

@@ -9,7 +9,7 @@
     "params": {
         "ResFileName": {
             "type": "STRING",
-            "value": "mpc_sub_result"
+            "value": "data/result/mpc_sub_result.csv"
         },
         "Expr": {
             "type": "STRING",

example/psi_ecdh_task_conf.json

@@ -8,12 +8,14 @@
   },
   "params": {
     "clientIndex": {
+      "description": "selected columns index for client dataset",
       "type": "INT32",
       "value": [0]
     },
     "serverIndex": {
+      "description": "selected columns index for server dataset",
       "type": "INT32",
-      "value": [1]
+      "value": [0]
     },
     "psiType": {
       "description": "availabe value: [INTERSECTION = 0; DIFFERENCE = 1;]",
@@ -26,14 +28,17 @@
       "value": 0
     },
     "outputFullFilename": {
+      "description": "path for client save intersection result",
       "type": "STRING",
       "value": "data/result/psi_result.csv"
     },
     "sync_result_to_server": {
+      "description": "whether client sync result to server or not. 1: true, 0: false",
       "type": "INT32",
       "value": 1
     },
     "server_outputFullFilname": {
+      "description": "path for server save intersection result",
       "type": "STRING",
       "value": "data/result/server/psi_result.csv"
     }
@@ -45,6 +50,5 @@
     "SERVER": {
       "SERVER": "psi_server_data"
     }
-  },
-  "input_datasets": []
+  }
 }

example/psi_kkrt_task_conf.json

@@ -8,10 +8,12 @@
   },
   "params": {
     "clientIndex": {
+      "description": "selected columns index for client dataset",
       "type": "INT32",
       "value": [0]
     },
     "serverIndex": {
+      "description": "selected columns index for server dataset",
       "type": "INT32",
       "value": [0]
     },
@@ -26,14 +28,17 @@
       "value": 1
     },
     "outputFullFilename": {
+      "description": "path for client save intersection result",
       "type": "STRING",
       "value": "data/result/psi_result.csv"
     },
     "sync_result_to_server": {
+      "description": "whether client sync result to server or not. 1: true, 0: false",
       "type": "INT32",
       "value": 1
     },
     "server_outputFullFilname": {
+      "description": "path for server save intersection result",
       "type": "STRING",
       "value": "data/result/server/psi_result.csv"
     }
@@ -45,6 +50,5 @@
     "SERVER": {
       "SERVER": "psi_server_data"
     }
-  },
-  "input_datasets": []
+  }
 }

src/primihub/algorithm/arithmetic.cc

@@ -400,15 +400,17 @@ template <Decimal Dbit> int ArithmeticExecutor<Dbit>::saveModel(void) {
         std::make_shared<arrow::Schema>(schema_vector_bool), {array});
   std::shared_ptr<DataDriver> driver =
       DataDirverFactory::getDriver("CSV", dataset_service_->getNodeletAddr());
-  std::shared_ptr<CSVDriver> csv_driver =
-      std::dynamic_pointer_cast<CSVDriver>(driver);
-
-  std::string filepath = "data/" + res_name_ + ".csv";
-  int ret = 0;
-  if (col_and_val_double.size() != 0)
-    ret = csv_driver->write(table, filepath);
-  else
-    ret = csv_driver->write(table, filepath);
+  // std::shared_ptr<CSVDriver> csv_driver =
+  //     std::dynamic_pointer_cast<CSVDriver>(driver);
+  auto& filepath = res_name_;
+  auto data_cursor = driver->initCursor(filepath);
+  auto dataset = std::make_shared<primihub::Dataset>(table, driver);
+  // int ret = 0;
+  // if (col_and_val_double.size() != 0)
+  //   ret = csv_driver->write(table, filepath);
+  // else
+  //   ret = csv_driver->write(table, filepath);
+  int ret = data_cursor->write(dataset);
   if (ret != 0) {
     LOG(ERROR) << "Save res to file " << filepath << " failed.";
     return -1;

src/primihub/algorithm/missing_val_processing.cc

@@ -356,8 +356,7 @@ int MissingProcess::loadParams(primihub::rpc::Task &task) {
   doc_ds.Swap(doc_iter->value);
 
   Value &vals = doc_ds["columns"];
-  for (Value::ConstMemberIterator iter = vals.MemberBegin();
-       iter != vals.MemberEnd(); iter++) {
+  for (auto iter = vals.MemberBegin(); iter != vals.MemberEnd(); iter++) {
     std::string col_name = iter->name.GetString();
     uint32_t col_dtype = iter->value.GetInt();
     col_and_dtype_.insert(std::make_pair(col_name, col_dtype));
@@ -365,8 +364,15 @@ int MissingProcess::loadParams(primihub::rpc::Task &task) {
               << iter->value.GetInt() << ".";
   }
   new_dataset_id_ = doc_ds["newDataSetId"].GetString();
-  LOG(INFO) << "New id of new dataset is " << new_dataset_id_ << ".";
+  if (doc_ds.HasMember("newDataPath") && doc_ds["newDataPath"].IsString()) {
+    new_dataset_path_ = doc_ds["newDataPath"].GetString();
+  } else {
+    new_dataset_path_ = "./" + new_dataset_id_ + ".csv";
+    LOG(WARNING) << "using default path: " << new_dataset_path_;
+  }
 
+  LOG(INFO) << "New id of new dataset is " << new_dataset_id_ << ". "
+      << "new dataset path: " << new_dataset_path_;
   return 0;
 }
 
@@ -1238,11 +1244,11 @@ int MissingProcess::finishPartyComm(void) {
 
 int MissingProcess::saveModel(void) {
   std::vector<std::string> str_vec;
-  std::string delimiter = ".";
-  int pos = data_file_path_.rfind(delimiter);
-
-  std::string new_path = data_file_path_.substr(0, pos) + "_missing.csv";
+  // std::string delimiter = ".";
+  // int pos = data_file_path_.rfind(delimiter);
 
+  // std::string new_path = data_file_path_.substr(0, pos) + "_missing.csv";
+  auto& new_path = new_dataset_path_;
   std::shared_ptr<DataDriver> driver =
       DataDirverFactory::getDriver("CSV", dataset_service_->getNodeletAddr());
 

src/primihub/algorithm/missing_val_processing.h

@@ -101,6 +101,7 @@ private:
   uint32_t party_id_;
 
   std::string new_dataset_id_;
+  std::string new_dataset_path_;
   std::string platform_type_ = "";
 
   template <class T>

src/primihub/data_store/csv/csv_driver.cc

@@ -25,6 +25,7 @@
 #include "src/primihub/data_store/csv/csv_driver.h"
 #include "src/primihub/data_store/driver.h"
 #include "src/primihub/util/util.h"
+#include "src/primihub/util/file_util.h"
 
 namespace primihub {
 namespace csv {
@@ -58,7 +59,7 @@ std::shared_ptr<arrow::Table> ReadCSVFile(const std::string& file_path,
         << "detail: " << maybe_table.status();
     return nullptr;
   }
-  
+
   return *maybe_table;
 }
 }  // namespace csv
@@ -244,6 +245,12 @@ std::shared_ptr<Dataset> CSVCursor::read(int64_t offset, int64_t limit) {
 }
 
 int CSVCursor::write(std::shared_ptr<Dataset> dataset) {
+  // check existence of file directory
+  auto ret = ValidateDir(this->filePath);
+  if (ret != 0) {
+    LOG(ERROR) << "something wrong with operatating file path: " << this->filePath;
+    return -1;
+  }
   // write Dataset to csv file
   auto result = arrow::io::FileOutputStream::Open(this->filePath);
   if (!result.ok()) {
@@ -349,7 +356,7 @@ std::unique_ptr<Cursor> CSVDriver::read() {
     if (arrow_data == nullptr) {
       return nullptr;
     }
-    
+
     std::vector<FieldType> fileds;
     auto arrow_fileds = arrow_data->schema()->fields();
     for (const auto& field : arrow_fileds) {
@@ -388,7 +395,11 @@ std::unique_ptr<Cursor> CSVDriver::initCursor(const std::string &filePath) {
 // FIXME to be deleted write file in driver directly.
 int CSVDriver::write(std::shared_ptr<arrow::Table> table,
                      const std::string& filePath) {
-  filePath_ = filePath;
+  auto ret = ValidateDir(filePath);
+  if (ret != 0) {
+    LOG(ERROR) << "something wrong with operatating file path: " << filePath;
+    return -1;
+  }
   auto result = arrow::io::FileOutputStream::Open(filePath);
   if (!result.ok()) {
     LOG(ERROR) << "Open file " << filePath << " failed. " << result.status();

src/primihub/task/semantic/mpc_task.cc

@@ -165,7 +165,11 @@ int MPCTask::execute() {
       }
 
       algorithm_->finishPartyComm();
-      algorithm_->saveModel();
+      ret = algorithm_->saveModel();
+      if (ret != 0) {
+        LOG(ERROR) << "saveModel failed.";
+        break;
+      }
     } while (0);
   } catch (std::exception &e) {
     LOG(ERROR) << e.what();