[migration] Gets entire directory tree in one SQL query

migration/gRPC/EosImportDirs.cpp

@@ -83,10 +83,8 @@ EosImportDirs::EosImportDirs(const std::string &configfile, std::string topLevel
 }
 
 
-void EosImportDirs::select(std::string tld)
+void EosImportDirs::select()
 {
-  if(tld.empty()) tld = m_topLevelPath + '%';
-
   const std::string sql =
     "SELECT "
     "  FILEID,"
@@ -104,7 +102,36 @@ void EosImportDirs::select(std::string tld)
     "ORDER BY "
     "  DEPTH ASC";
   m_selectDirsDbStmt = m_dbConn.createStmt(sql);
-  m_selectDirsDbStmt.bindString(":TOP_LEVEL_PATH", tld);
+  m_selectDirsDbStmt.bindString(":TOP_LEVEL_PATH", m_topLevelPath + '%');
+  std::cerr << "Executing SELECT query against " << m_tableName << " table...";
+  m_selectDirsDbRset = m_selectDirsDbStmt.executeQuery();
+  std::cerr << "done." << std::endl;
+}
+
+
+void EosImportDirs::select(const std::list<std::string> &dirs)
+{
+  std::string sql =
+    "SELECT "
+    "  FILEID,"
+    "  PATH,"
+    "  DISK_UID,"
+    "  DISK_GID,"
+    "  FILEMODE,"
+    "  CTIME,"
+    "  MTIME,"
+    "  CLASSID,"
+    "  DEPTH "
+    "FROM " + m_tableName;
+
+  std::string connector = " WHERE ";
+  for(auto &dir : dirs) {
+    sql += connector + "PATH='" + dir + "'";
+    connector = " OR ";
+  }
+  sql += " ORDER BY DEPTH ASC";
+
+  m_selectDirsDbStmt = m_dbConn.createStmt(sql);
   std::cerr << "Executing SELECT query against " << m_tableName << " table...";
   m_selectDirsDbRset = m_selectDirsDbStmt.executeQuery();
   std::cerr << "done." << std::endl;
@@ -305,10 +332,17 @@ void EosImportDirs::forgetFailures()
 void EosImportDirs::checkRootDirExists()
 {
   auto tld = eos::client::manglePathname(m_castor_prefix, m_eos_prefix, m_topLevelPath);
-  std::string &path = tld.pathname;
-  auto tld_pb = m_eosgrpc->GetMD(eos::rpc::CONTAINER, 0, path);
+  auto tld_pb = m_eosgrpc->GetMD(eos::rpc::CONTAINER, 0, tld.pathname);
+
+  // Root directory for the import exists, nothing more to do
+  if(tld_pb.cmd().id() != 0) return;
+
+  if(m_topLevelPath[0] != '/') {
+    throw std::runtime_error("Parent directory for import must begin with /");
+  }
 
-  // If the root directory does not exist, try to create the tree up to the root
+  // Try to create the tree up to the root
+  std::string path = m_topLevelPath;
   std::list<std::string> dirsToCreate;
   while(tld_pb.cmd().id() == 0) {
     // Strip off any trailing slashes
@@ -319,10 +353,14 @@ void EosImportDirs::checkRootDirExists()
     path.resize(path.find_last_of('/'));
     tld_pb = m_eosgrpc->GetMD(eos::rpc::CONTAINER, 0, path);
   }
-  for(auto &dir : dirsToCreate) {
-    std::cerr << "Creating " << dir << ":" << std::endl;
-    select(dir);
-    processBatch();
+  std::cerr << "Creating parent directory " << m_topLevelPath << "..." << std::endl;
+  select(dirsToCreate);
+  processBatch();
+
+  // Check the root directory was created
+  tld_pb = m_eosgrpc->GetMD(eos::rpc::CONTAINER, 0, tld.pathname);
+  if(tld_pb.cmd().id() == 0) {
+    throw std::runtime_error("Could not create parent directory for import");
   }
 }
 

migration/gRPC/EosImportDirs.hpp

@@ -33,10 +33,11 @@ public:
   void listFailureModes();
   void listFailures();
   void forgetFailures();
-  void select(std::string tld = "");
+  void select();
   bool processBatch();
 
 private:
+  void select(const std::list<std::string> &dirs);
   void saveFailedDirs(const std::vector<eos::rpc::ContainerMdProto> &dirs, const eos::rpc::InsertReply &replies);
   bool compareMD(const eos::rpc::ContainerMdProto &dir);
   std::vector<eos::rpc::ContainerMdProto> getNextBatch();