Merge pull request #4000 from dmitry-ganyushin/campaign-json-to-sqlite3

Implementing sqlite database instead of json file. Writing part.

source/adios2/engine/campaign/CampaignManager.cpp

@@ -17,6 +17,7 @@
 #include <iostream>
 
 #include <nlohmann_json.hpp>
+#include <sqlite3.h>
 
 namespace adios2
 {
@@ -25,20 +26,54 @@ namespace core
 namespace engine
 {
 
-static std::string CMapToJson(const CampaignRecordMap &cmap, const int rank, const std::string name)
+int CMapToSqlite(const CampaignRecordMap &cmap, const int rank, std::string name)
 {
-    nlohmann::json j = nlohmann::json::array();
+    sqlite3 *db;
+    int rc;
+    char *zErrMsg = 0;
+    std::string sqlcmd;
+    std::string db_name = name + ".db";
+    rc = sqlite3_open(db_name.c_str(), &db);
+
+    if (rc != SQLITE_OK)
+    {
+        std::cout << "SQL error: " << zErrMsg << std::endl;
+        std::string m(zErrMsg);
+        helper::Throw<std::invalid_argument>("Engine", "CampaignReader", "WriteCampaignData",
+                                             "SQL error on writing records:");
+        sqlite3_free(zErrMsg);
+    }
+    sqlcmd = "CREATE TABLE bpfiles (name);";
+    rc = sqlite3_exec(db, sqlcmd.c_str(), 0, 0, &zErrMsg);
+    if (rc != SQLITE_OK)
+    {
+        std::cout << "SQL error: " << zErrMsg << std::endl;
+        std::string m(zErrMsg);
+        helper::Throw<std::invalid_argument>("Engine", "CampaignReader", "WriteCampaignData",
+                                             "SQL error on writing records:");
+        sqlite3_free(zErrMsg);
+    }
+
+    size_t rowid = 1000;
     for (auto &r : cmap)
     {
-        nlohmann::json c = nlohmann::json{{"name", r.first},
-                                          {"varying_deltas", r.second.varying_deltas},
-                                          {"delta_step", r.second.delta_step},
-                                          {"delta_time", r.second.delta_time},
-                                          {"steps", r.second.steps},
-                                          {"times", r.second.times}};
-        j.push_back(c);
+        sqlcmd = "INSERT INTO bpfiles (rowid, name)\n";
+        sqlcmd += "VALUES(" + std::to_string(rowid) + "," + "'" + r.first + "'" + ");";
+        rowid++;
+        rc = sqlite3_exec(db, sqlcmd.c_str(), 0, 0, &zErrMsg);
+        if (rc != SQLITE_OK)
+        {
+            std::cout << "SQL error: " << zErrMsg << std::endl;
+            std::string m(zErrMsg);
+            helper::Throw<std::invalid_argument>("Engine", "CampaignReader", "WriteCampaignData",
+                                                 "SQL error on writing records:");
+            sqlite3_free(zErrMsg);
+        }
     }
-    return nlohmann::to_string(j);
+
+    sqlite3_close(db);
+
+    return 0;
 }
 
 CampaignManager::CampaignManager(adios2::helper::Comm &comm)
@@ -65,7 +100,7 @@ CampaignManager::~CampaignManager()
 
 void CampaignManager::Open(const std::string &name)
 {
-    m_Name = m_CampaignDir + "/" + name + "_" + std::to_string(m_WriterRank) + ".json";
+    m_Name = m_CampaignDir + "/" + name + "_" + std::to_string(m_WriterRank);
     if (m_Verbosity == 5)
     {
         std::cout << "Campaign Manager " << m_WriterRank << " Open(" << m_Name << ")\n";
@@ -118,11 +153,7 @@ void CampaignManager::Close()
 {
     if (!cmap.empty())
     {
-        m_Output.open(m_Name, std::ofstream::out);
-        m_Opened = true;
-        m_Output << std::setw(4) << CMapToJson(cmap, m_WriterRank, m_Name) << std::endl;
-        m_Output.close();
-        m_Opened = false;
+        CMapToSqlite(cmap, m_WriterRank, m_Name);
     }
 }
 

source/utils/adios_campaign_manager/adios2_campaign_manager.py

@@ -222,10 +222,10 @@ def Update(args: dict, cur: sqlite3.Cursor):
     dirID = curDir.lastrowid
     con.commit()
 
-    jsonlist = MergeJsonFiles(jsonFileList)
+    db_list = MergeDBFiles(dbFileList)
 
     # print(f"Merged json = {jsonlist}")
-    ProcessJsonFile(args, jsonlist, cur, hostID, dirID)
+    ProcessJsonFile(args, db_list, cur, hostID, dirID)
 
     con.commit()
 
@@ -250,11 +250,24 @@ def Create(args: dict, cur: sqlite3.Cursor):
     Update(args, cur)
 
 
-def MergeJsonFiles(jsonfiles: list):
+def MergeDBFiles(dbfiles: list):
+    # read db files here
     result = list()
-    for f1 in jsonfiles:
-        with open(f1, 'r') as infile:
-            result.extend(json.load(infile))
+    for f1 in dbfiles:
+        try:
+            con = sqlite3.connect(f1)
+        except sqlite3.Error as e:
+            print(e)
+
+        cur = con.cursor()
+        try:
+            cur.execute("select  * from bpfiles")
+        except sqlite3.Error as e:
+            print(e)
+        record = cur.fetchall()
+        for item in record:
+            result.append({"name": item[0]})
+        cur.close()
     return result
 
 
@@ -291,8 +304,8 @@ def MergeJsonFiles(jsonfiles: list):
     else:
         CheckLocalCampaignDir(args)
         # List the local campaign directory
-        jsonFileList = glob.glob(args.LocalCampaignDir + '/*.json')
-        if len(jsonFileList) == 0:
+        dbFileList = glob.glob(args.LocalCampaignDir + '/*.db')
+        if len(dbFileList) == 0:
             print("There are no campaign data files in  " + args.LocalCampaignDir)
             exit(2)