<NME> swissknife_migrate.cc
<BEF> /**
 * This file is part of the CernVM File System.
 *
 * Careful: any real schema migration as of now requires taking care of
 * hash algorithm
 */

#include "swissknife_migrate.h"

#include <sys/resource.h>
#include <unistd.h>

#include "catalog_rw.h"
#include "catalog_sql.h"
#include "catalog_virtual.h"
#include "compression.h"
#include "crypto/hash.h"
#include "swissknife_history.h"
#include "util/concurrency.h"
#include "util/logging.h"

using namespace std;  // NOLINT

namespace swissknife {

catalog::DirectoryEntry  CommandMigrate::nested_catalog_marker_;

CommandMigrate::CommandMigrate() :
  file_descriptor_limit_(8192),
  catalog_count_(0),
  has_committed_new_revision_(false),
  uid_(0),
  gid_(0),
  root_catalog_(NULL)
{
  atomic_init32(&catalogs_processed_);
}


ParameterList CommandMigrate::GetParams() const {
  ParameterList r;
  r.push_back(Parameter::Mandatory('v',
    "migration base version ( 2.0.x | 2.1.7 | chown | hardlink | bulkhash | "
    "stats)"));
  r.push_back(Parameter::Mandatory('r',
    "repository URL (absolute local path or remote URL)"));
  r.push_back(Parameter::Mandatory('u', "upstream definition string"));
  r.push_back(Parameter::Mandatory('o', "manifest output file"));
  r.push_back(Parameter::Mandatory('t',
    "temporary directory for catalog decompress"));
  r.push_back(Parameter::Optional('p',
    "user id to be used for this repository"));
  r.push_back(Parameter::Optional('g',
    "group id to be used for this repository"));
  r.push_back(Parameter::Optional('n', "fully qualified repository name"));
  r.push_back(Parameter::Optional('h', "root hash (other than trunk)"));
  r.push_back(Parameter::Optional('k', "repository master key(s)"));
  r.push_back(Parameter::Optional('i', "UID map for chown"));
  r.push_back(Parameter::Optional('j', "GID map for chown"));
  r.push_back(Parameter::Optional('@', "proxy url"));
  r.push_back(Parameter::Switch('f', "fix nested catalog transition points"));
  r.push_back(Parameter::Switch('l', "disable linkcount analysis of files"));
  r.push_back(Parameter::Switch('s',
    "enable collection of catalog statistics"));
  return r;
}


static void Error(const std::string &message) {
  LogCvmfs(kLogCatalog, kLogStderr, message.c_str());
}


static void Error(const std::string                     &message,
                  const CommandMigrate::PendingCatalog  *catalog) {
  const std::string err_msg = message + "\n"
                              "Catalog: " + catalog->root_path();
  Error(err_msg);
}


static void Error(const std::string                     &message,
                  const catalog::SqlCatalog             &statement,
                  const CommandMigrate::PendingCatalog  *catalog) {
  const std::string err_msg =
    message + "\n"
    "SQLite: " + StringifyInt(statement.GetLastError()) +
    " - " + statement.GetLastErrorMsg();
  Error(err_msg, catalog);
}


int CommandMigrate::Main(const ArgumentList &args) {
  shash::Any manual_root_hash;
  const std::string &migration_base     = *args.find('v')->second;
  const std::string &repo_url           = *args.find('r')->second;
  const std::string &spooler            = *args.find('u')->second;
  const std::string &manifest_path      = *args.find('o')->second;
  const std::string &tmp_dir            = *args.find('t')->second;
  const std::string &uid                = (args.count('p') > 0)      ?
                                             *args.find('p')->second :
                                             "";
  const std::string &gid                = (args.count('g') > 0)      ?
                                             *args.find('g')->second :
                                             "";
  const std::string &repo_name          = (args.count('n') > 0)      ?
                                             *args.find('n')->second :
                                             "";
  const std::string &repo_keys          = (args.count('k') > 0)      ?
                                             *args.find('k')->second :
                                             "";
  const std::string &uid_map_path       = (args.count('i') > 0)      ?
                                             *args.find('i')->second :
                                             "";
  const std::string &gid_map_path       = (args.count('j') > 0)      ?
                                             *args.find('j')->second :
                                             "";
  const bool fix_transition_points      = (args.count('f') > 0);
  const bool analyze_file_linkcounts    = (args.count('l') == 0);
  const bool collect_catalog_statistics = (args.count('s') > 0);
  if (args.count('h') > 0) {
    manual_root_hash = shash::MkFromHexPtr(shash::HexPtr(
      *args.find('h')->second), shash::kSuffixCatalog);
  }

  // We might need a lot of file descriptors
  if (!RaiseFileDescriptorLimit()) {
    Error("Failed to raise file descriptor limits");
    return 2;
  }

  // Put SQLite into multithreaded mode
  if (!ConfigureSQLite()) {
    Error("Failed to preconfigure SQLite library");
    return 3;
  }

  // Create an upstream spooler
  temporary_directory_ = tmp_dir;
  const upload::SpoolerDefinition spooler_definition(spooler, shash::kSha1);
  spooler_ = upload::Spooler::Construct(spooler_definition);
  if (!spooler_.IsValid()) {
    Error("Failed to create upstream Spooler.");
    return 5;
  }
  spooler_->RegisterListener(&CommandMigrate::UploadCallback, this);

  // Load the full catalog hierarchy
  LogCvmfs(kLogCatalog, kLogStdout, "Loading current catalog tree...");

  catalog_loading_stopwatch_.Start();
  bool loading_successful = false;
  if (IsHttpUrl(repo_url)) {
    typedef HttpObjectFetcher<catalog::WritableCatalog> ObjectFetcher;

    const bool follow_redirects = false;
    const string proxy = (args.count('@') > 0) ? *args.find('@')->second : "";
    if (!this->InitDownloadManager(follow_redirects, proxy) ||
        !this->InitVerifyingSignatureManager(repo_keys)) {
      LogCvmfs(kLogCatalog, kLogStderr, "Failed to init repo connection");
      return 1;
    }

    ObjectFetcher fetcher(repo_name,
                          repo_url,
                          tmp_dir,
                          download_manager(),
                          signature_manager());

    loading_successful = LoadCatalogs(manual_root_hash, &fetcher);
  } else {
    typedef LocalObjectFetcher<catalog::WritableCatalog> ObjectFetcher;
    ObjectFetcher fetcher(repo_url, tmp_dir);
    loading_successful = LoadCatalogs(manual_root_hash, &fetcher);
  }
  catalog_loading_stopwatch_.Stop();

  if (!loading_successful) {
    Error("Failed to load catalog tree");
    return 4;
  }

  LogCvmfs(kLogCatalog, kLogStdout, "Loaded %d catalogs", catalog_count_);
  assert(root_catalog_ != NULL);

  // Do the actual migration step
  bool migration_succeeded = false;
  if (migration_base == "2.0.x") {
    if (!ReadPersona(uid, gid)) {
      return 1;
    }

    // Generate and upload a nested catalog marker
    if (!GenerateNestedCatalogMarkerChunk()) {
      Error("Failed to create a nested catalog marker.");
      return 6;
    }
    spooler_->WaitForUpload();

    // Configure the concurrent catalog migration facility
    MigrationWorker_20x::worker_context context(temporary_directory_,
                                                collect_catalog_statistics,
                                                fix_transition_points,
                                                analyze_file_linkcounts,
                                                uid_,
                                                gid_);
    migration_succeeded =
      DoMigrationAndCommit<MigrationWorker_20x>(manifest_path, &context);
  } else if (migration_base == "2.1.7") {
    MigrationWorker_217::worker_context context(temporary_directory_,
                                                collect_catalog_statistics);
    migration_succeeded =
      DoMigrationAndCommit<MigrationWorker_217>(manifest_path, &context);
  } else if (migration_base == "chown") {
    UidMap uid_map;
    GidMap gid_map;
    if (!ReadPersonaMaps(uid_map_path, gid_map_path, &uid_map, &gid_map)) {
      Error("Failed to read UID and/or GID map");
      return 1;
    }
    ChownMigrationWorker::worker_context context(temporary_directory_,
                                                 collect_catalog_statistics,
                                                 uid_map,
                                                 gid_map);
    migration_succeeded =
      DoMigrationAndCommit<ChownMigrationWorker>(manifest_path, &context);
  } else if (migration_base == "hardlink") {
    HardlinkRemovalMigrationWorker::worker_context
      context(temporary_directory_, collect_catalog_statistics);
    migration_succeeded =
      DoMigrationAndCommit<HardlinkRemovalMigrationWorker>(manifest_path,
                                                           &context);
  } else if (migration_base == "bulkhash") {
    BulkhashRemovalMigrationWorker::worker_context
      context(temporary_directory_, collect_catalog_statistics);
    migration_succeeded =
      DoMigrationAndCommit<BulkhashRemovalMigrationWorker>(manifest_path,
                                                           &context);
  } else if (migration_base == "stats") {
    StatsMigrationWorker::worker_context context(
      temporary_directory_, collect_catalog_statistics);
    migration_succeeded =
      DoMigrationAndCommit<StatsMigrationWorker>(manifest_path, &context);
  } else {
    const std::string err_msg = "Unknown migration base: " + migration_base;
    Error(err_msg);
    return 1;
  }

  // Check if everything went well
  if (!migration_succeeded) {
    Error("Migration failed!");
    return 5;
  }

  // Analyze collected statistics
  if (collect_catalog_statistics && has_committed_new_revision_) {
    LogCvmfs(kLogCatalog, kLogStdout, "\nCollected statistics results:");
    AnalyzeCatalogStatistics();
  }

  LogCvmfs(kLogCatalog, kLogStdout, "\nCatalog Migration succeeded");
  return 0;
}


bool CommandMigrate::ReadPersona(const std::string &uid,
                                   const std::string &gid) {
  if (uid.empty()) {
    Error("Please provide a user ID");
    return false;
  }
  if (gid.empty()) {
    Error("Please provide a group ID");
    return false;
  }

  uid_ = String2Int64(uid);
  gid_ = String2Int64(gid);
  return true;
}



bool CommandMigrate::ReadPersonaMaps(const std::string &uid_map_path,
                                     const std::string &gid_map_path,
                                           UidMap      *uid_map,
                                           GidMap      *gid_map) const {
  if (!uid_map->Read(uid_map_path) || !uid_map->IsValid()) {
    Error("Failed to read UID map");
    return false;
  }

  if (!gid_map->Read(gid_map_path) || !gid_map->IsValid()) {
    Error("Failed to read GID map");
    return false;
  }

  if (uid_map->RuleCount() == 0 && !uid_map->HasDefault()) {
    Error("UID map appears to be empty");
    return false;
  }

  if (gid_map->RuleCount() == 0 && !gid_map->HasDefault()) {
    Error("GID map appears to be empty");
    return false;
  }

  return true;
}


void CommandMigrate::UploadHistoryClosure(
  const upload::SpoolerResult &result,
  Future<shash::Any> *hash)
{
  assert(!result.IsChunked());
  if (result.return_code != 0) {
    LogCvmfs(kLogCvmfs, kLogStderr, "failed to upload history database (%d)",
             result.return_code);
    hash->Set(shash::Any());
  } else {
    hash->Set(result.content_hash);
  }
}


bool CommandMigrate::UpdateUndoTags(
  PendingCatalog *root_catalog,
  unsigned revision,
  time_t timestamp,
  shash::Any *history_hash)
{
  string filename_old = history_upstream_->filename();
  string filename_new = filename_old + ".new";
  bool retval = CopyPath2Path(filename_old, filename_new);
  if (!retval) return false;
  UniquePtr<history::SqliteHistory> history(
    history::SqliteHistory::OpenWritable(filename_new));
  history->TakeDatabaseFileOwnership();

  history::History::Tag tag_trunk;
  bool exists = history->GetByName(CommandTag::kHeadTag, &tag_trunk);
  if (exists) {
    retval = history->Remove(CommandTag::kHeadTag);
    if (!retval) return false;

    history::History::Tag tag_trunk_previous = tag_trunk;
    tag_trunk_previous.name = CommandTag::kPreviousHeadTag;
    tag_trunk_previous.description = CommandTag::kPreviousHeadTagDescription;
    history->Remove(CommandTag::kPreviousHeadTag);

    tag_trunk.root_hash = root_catalog->new_catalog_hash;
    tag_trunk.size = root_catalog->new_catalog_size;
    tag_trunk.revision = root_catalog->new_catalog_size;
    tag_trunk.revision = revision;
    tag_trunk.timestamp = timestamp;

    retval = history->Insert(tag_trunk_previous);
    if (!retval) return false;
    retval = history->Insert(tag_trunk);
    if (!retval) return false;
  }

  history->SetPreviousRevision(manifest_upstream_->history());
  history->DropDatabaseFileOwnership();
  history.Destroy();

  Future<shash::Any> history_hash_new;
  upload::Spooler::CallbackPtr callback = spooler_->RegisterListener(
    &CommandMigrate::UploadHistoryClosure, this, &history_hash_new);
  spooler_->ProcessHistory(filename_new);
  spooler_->WaitForUpload();
  spooler_->UnregisterListener(callback);
  unlink(filename_new.c_str());
  *history_hash = history_hash_new.Get();
  if (history_hash->IsNull()) {
    Error("failed to upload tag database");
    return false;
  }

  return true;
}


template <class MigratorT>
bool CommandMigrate::DoMigrationAndCommit(
  const std::string                   &manifest_path,
  typename MigratorT::worker_context  *context
) {
  // Create a concurrent migration context for catalog migration
  const unsigned int cpus = GetNumberOfCpuCores();
  ConcurrentWorkers<MigratorT> concurrent_migration(cpus, cpus * 10, context);

  if (!concurrent_migration.Initialize()) {
    Error("Failed to initialize worker migration system.");
    return false;
  }
  concurrent_migration.RegisterListener(&CommandMigrate::MigrationCallback,
                                         this);

  // Migrate catalogs recursively (starting with the deepest nested catalogs)
  LogCvmfs(kLogCatalog, kLogStdout, "\nMigrating catalogs...");
  PendingCatalog *root_catalog = new PendingCatalog(root_catalog_);
  migration_stopwatch_.Start();
  ConvertCatalogsRecursively(root_catalog, &concurrent_migration);
  concurrent_migration.WaitForEmptyQueue();
  spooler_->WaitForUpload();
  spooler_->UnregisterListeners();
  migration_stopwatch_.Stop();

  // check for possible errors during the migration process
  const unsigned int errors = concurrent_migration.GetNumberOfFailedJobs() +
                              spooler_->GetNumberOfErrors();
  LogCvmfs(kLogCatalog, kLogStdout,
           "Catalog Migration finished with %d errors.", errors);
  if (errors > 0) {
    LogCvmfs(kLogCatalog, kLogStdout,
             "\nCatalog Migration produced errors\nAborting...");
    return false;
  }

  if (root_catalog->was_updated.Get()) {
    LogCvmfs(kLogCatalog, kLogStdout,
             "\nCommitting migrated repository revision...");
    manifest::Manifest manifest = *manifest_upstream_;
    manifest.set_catalog_hash(root_catalog->new_catalog_hash);
    manifest.set_catalog_size(root_catalog->new_catalog_size);
    manifest.set_root_path(root_catalog->root_path());
    const catalog::Catalog* new_catalog = (root_catalog->HasNew())
                                          ? root_catalog->new_catalog
                                          : root_catalog->old_catalog;
    manifest.set_ttl(new_catalog->GetTTL());
    manifest.set_revision(new_catalog->GetRevision());

    // Commit the new (migrated) repository revision...
    if (history_upstream_.IsValid()) {
      shash::Any history_hash(manifest_upstream_->history());
      LogCvmfs(kLogCatalog, kLogStdout | kLogNoLinebreak,
               "Updating repository tag database... ");
      if (!UpdateUndoTags(root_catalog,
                          new_catalog->GetRevision(),
                          new_catalog->GetLastModified(),
                          &history_hash))
      {
        Error("Updateing tag database failed.\nAborting...");
        return false;
      }
      manifest.set_history(history_hash);
      LogCvmfs(kLogCvmfs, kLogStdout, "%s", history_hash.ToString().c_str());
    }

    if (!manifest.Export(manifest_path)) {
      Error("Manifest export failed.\nAborting...");
      return false;
    }
    has_committed_new_revision_ = true;
  } else {
    LogCvmfs(kLogCatalog, kLogStdout,
             "\nNo catalogs migrated, skipping the commit...");
  }

  // Get rid of the open root catalog
  delete root_catalog;

  return true;
}


void CommandMigrate::CatalogCallback(
                   const CatalogTraversalData<catalog::WritableCatalog> &data) {
  std::string tree_indent;
  std::string hash_string;
  std::string path;

  for (unsigned int i = 1; i < data.tree_level; ++i) {
    tree_indent += "\u2502  ";
  }

  if (data.tree_level > 0) {
    tree_indent += "\u251C\u2500 ";
  }

  hash_string = data.catalog_hash.ToString();

  path = data.catalog->mountpoint().ToString();
  if (path.empty()) {
    path = "/";
    root_catalog_ = data.catalog;
  }

  LogCvmfs(kLogCatalog, kLogStdout, "%s%s %s",
    tree_indent.c_str(),
    hash_string.c_str(),
    path.c_str());

  ++catalog_count_;
}


void CommandMigrate::MigrationCallback(PendingCatalog *const &data) {
  // Check if the migration of the catalog was successful
  if (!data->success) {
    Error("Catalog migration failed! Aborting...");
    exit(1);
    return;
  }

  if (!data->HasChanges()) {
    PrintStatusMessage(data, data->GetOldContentHash(), "preserved");
    data->was_updated.Set(false);
    return;
  }

  const string &path = (data->HasNew()) ? data->new_catalog->database_path()
                                        : data->old_catalog->database_path();

  // Save the processed catalog in the pending map
  {
    LockGuard<PendingCatalogMap> guard(&pending_catalogs_);
    assert(pending_catalogs_.find(path) == pending_catalogs_.end());
    pending_catalogs_[path] = data;
  }
  catalog_statistics_list_.Insert(data->statistics);

  // check the size of the uncompressed catalog file
  size_t new_catalog_size = GetFileSize(path);
  if (new_catalog_size <= 0) {
    Error("Failed to get uncompressed file size of catalog!", data);
    exit(2);
    return;
  }
  data->new_catalog_size = new_catalog_size;

  // Schedule the compression and upload of the catalog
  spooler_->ProcessCatalog(path);
}


void CommandMigrate::UploadCallback(const upload::SpoolerResult &result) {
  const string &path = result.local_path;

  // Check if the upload was successful
  if (result.return_code != 0) {
    Error("Failed to upload file " + path + "\nAborting...");
    exit(2);
    return;
  }
  assert(result.file_chunks.size() == 0);

  // Remove the just uploaded file
  unlink(path.c_str());

  // Uploaded nested catalog marker... generate and cache DirectoryEntry for it
  if (path == nested_catalog_marker_tmp_path_) {
    CreateNestedCatalogMarkerDirent(result.content_hash);
    return;
  } else {
    // Find the catalog path in the pending catalogs and remove it from the list
    PendingCatalog *catalog;
    {
      LockGuard<PendingCatalogMap> guard(&pending_catalogs_);
      PendingCatalogMap::iterator i = pending_catalogs_.find(path);
      assert(i != pending_catalogs_.end());
      catalog = const_cast<PendingCatalog*>(i->second);
      pending_catalogs_.erase(i);
    }

    PrintStatusMessage(catalog, result.content_hash, "migrated and uploaded");

    // The catalog is completely processed... fill the content_hash to allow the
    // processing of parent catalogs (Notified by 'was_updated'-future)
    // NOTE: From now on, this PendingCatalog structure could be deleted and
    //       should not be used anymore!
    catalog->new_catalog_hash = result.content_hash;
    catalog->was_updated.Set(true);
  }
}


void CommandMigrate::PrintStatusMessage(const PendingCatalog *catalog,
                                        const shash::Any     &content_hash,
                                        const std::string    &message) {
  atomic_inc32(&catalogs_processed_);
  const unsigned int processed = (atomic_read32(&catalogs_processed_) * 100) /
                                  catalog_count_;
  LogCvmfs(kLogCatalog, kLogStdout, "[%d%%] %s %sC %s",
           processed,
           message.c_str(),
           content_hash.ToString().c_str(),
           catalog->root_path().c_str());
}


template <class MigratorT>
void CommandMigrate::ConvertCatalogsRecursively(PendingCatalog *catalog,
                                                MigratorT       *migrator) {
  // First migrate all nested catalogs (depth first traversal)
  const catalog::CatalogList nested_catalogs =
    catalog->old_catalog->GetChildren();
  catalog::CatalogList::const_iterator i    = nested_catalogs.begin();
  catalog::CatalogList::const_iterator iend = nested_catalogs.end();
  catalog->nested_catalogs.reserve(nested_catalogs.size());
  for (; i != iend; ++i) {
    PendingCatalog *new_nested = new PendingCatalog(*i);
    catalog->nested_catalogs.push_back(new_nested);
    ConvertCatalogsRecursively(new_nested, migrator);
  }

  // Migrate this catalog referencing all its (already migrated) children
  migrator->Schedule(catalog);
}


bool CommandMigrate::RaiseFileDescriptorLimit() const {
  struct rlimit rpl;
  memset(&rpl, 0, sizeof(rpl));
  getrlimit(RLIMIT_NOFILE, &rpl);
  if (rpl.rlim_cur < file_descriptor_limit_) {
    if (rpl.rlim_max < file_descriptor_limit_)
      rpl.rlim_max = file_descriptor_limit_;
    rpl.rlim_cur = file_descriptor_limit_;
    const bool retval = setrlimit(RLIMIT_NOFILE, &rpl);
    if (retval != 0) {
      return false;
    }
  }
  return true;
}


bool CommandMigrate::ConfigureSQLite() const {
  int retval = sqlite3_config(SQLITE_CONFIG_MULTITHREAD);
  return (retval == SQLITE_OK);
}


void CommandMigrate::AnalyzeCatalogStatistics() const {
  const unsigned int number_of_catalogs = catalog_statistics_list_.size();
  unsigned int       aggregated_entry_count = 0;
  unsigned int       aggregated_max_row_id = 0;
  unsigned int       aggregated_hardlink_count = 0;
  unsigned int       aggregated_linkcounts = 0;
  double             aggregated_migration_time = 0.0;

  CatalogStatisticsList::const_iterator i    = catalog_statistics_list_.begin();
  CatalogStatisticsList::const_iterator iend = catalog_statistics_list_.end();
  for (; i != iend; ++i) {
    aggregated_entry_count    += i->entry_count;
    aggregated_max_row_id     += i->max_row_id;
    aggregated_hardlink_count += i->hardlink_group_count;
    aggregated_linkcounts     += i->aggregated_linkcounts;
    aggregated_migration_time += i->migration_time;
  }

  // Inode quantization
  assert(aggregated_max_row_id > 0);
  const unsigned int unused_inodes =
                                 aggregated_max_row_id - aggregated_entry_count;
  const float ratio =
    (static_cast<float>(unused_inodes) /
     static_cast<float>(aggregated_max_row_id)) * 100.0f;
  LogCvmfs(kLogCatalog, kLogStdout, "Actual Entries:                %d\n"
                                    "Allocated Inodes:              %d\n"
                                    "  Unused Inodes:               %d\n"
                                    "  Percentage of wasted Inodes: %.1f%%\n",
           aggregated_entry_count, aggregated_max_row_id, unused_inodes, ratio);

  // Hardlink statistics
  const float average_linkcount = (aggregated_hardlink_count > 0)
                                  ? aggregated_linkcounts /
                                    aggregated_hardlink_count
                                  : 0.0f;
  LogCvmfs(kLogCatalog, kLogStdout, "Generated Hardlink Groups:     %d\n"
                                    "Average Linkcount per Group:   %.1f\n",
           aggregated_hardlink_count, average_linkcount);

  // Performance measures
  const double average_migration_time =
    aggregated_migration_time / static_cast<double>(number_of_catalogs);
  LogCvmfs(kLogCatalog, kLogStdout, "Catalog Loading Time:          %.2fs\n"
                                    "Average Migration Time:        %.2fs\n"
                                    "Overall Migration Time:        %.2fs\n"
                                    "Aggregated Migration Time:     %.2fs\n",
           catalog_loading_stopwatch_.GetTime(),
           average_migration_time,
           migration_stopwatch_.GetTime(),
           aggregated_migration_time);
}


CommandMigrate::PendingCatalog::~PendingCatalog() {
  delete old_catalog;
  old_catalog = NULL;

  if (new_catalog != NULL) {
    delete new_catalog;
    new_catalog = NULL;
  }
}


template<class DerivedT>
CommandMigrate::AbstractMigrationWorker<DerivedT>::AbstractMigrationWorker(
  const worker_context *context)
  : temporary_directory_(context->temporary_directory)
  , collect_catalog_statistics_(context->collect_catalog_statistics)
{ }


template<class DerivedT>
CommandMigrate::AbstractMigrationWorker<DerivedT>::~AbstractMigrationWorker() {}


template<class DerivedT>
void CommandMigrate::AbstractMigrationWorker<DerivedT>::operator()(
                                                    const expected_data &data) {
  migration_stopwatch_.Start();
  const bool success = static_cast<DerivedT*>(this)->RunMigration(data) &&
                       UpdateNestedCatalogReferences(data) &&
                       UpdateCatalogMetadata(data)         &&
                       CollectAndAggregateStatistics(data) &&
                       CleanupNestedCatalogs(data);
  data->success = success;
  migration_stopwatch_.Stop();

  data->statistics.migration_time = migration_stopwatch_.GetTime();
  migration_stopwatch_.Reset();

  // Note: MigrationCallback() will take care of the result...
  if (success) {
    ConcurrentWorker<DerivedT>::master()->JobSuccessful(data);
  } else {
    ConcurrentWorker<DerivedT>::master()->JobFailed(data);
  }
}


template<class DerivedT>
bool CommandMigrate::AbstractMigrationWorker<DerivedT>::
     UpdateNestedCatalogReferences(PendingCatalog *data) const
{
  const catalog::Catalog *new_catalog =
    (data->HasNew()) ? data->new_catalog : data->old_catalog;
  const catalog::CatalogDatabase &writable = new_catalog->database();

  catalog::SqlCatalog add_nested_catalog(writable,
    "INSERT OR REPLACE INTO nested_catalogs (path,   sha1,  size) "
    "                VALUES                 (:path, :sha1, :size);");

  // go through all nested catalogs and update their references (we are curently
  // in their parent catalog)
  // Note: we might need to wait for the nested catalog to be fully processed.
  PendingCatalogList::const_iterator i    = data->nested_catalogs.begin();
  PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    PendingCatalog    *nested_catalog  = *i;

    if (!nested_catalog->was_updated.Get()) {
      continue;
    }

    const std::string &root_path    = nested_catalog->root_path();
    const shash::Any   catalog_hash = nested_catalog->new_catalog_hash;
    const size_t       catalog_size = nested_catalog->new_catalog_size;

    // insert the updated nested catalog reference into the new catalog
    const bool retval =
      add_nested_catalog.BindText(1, root_path)               &&
      add_nested_catalog.BindText(2, catalog_hash.ToString()) &&
      add_nested_catalog.BindInt64(3, catalog_size)           &&
      add_nested_catalog.Execute();
    if (!retval) {
      Error("Failed to add nested catalog link", add_nested_catalog, data);
      return false;
    }
    add_nested_catalog.Reset();
  }

  return true;
}


template<class DerivedT>
bool CommandMigrate::AbstractMigrationWorker<DerivedT>::
     UpdateCatalogMetadata(PendingCatalog *data) const
{
  if (!data->HasChanges()) {
    return true;
  }

  catalog::WritableCatalog *catalog =
    (data->HasNew()) ? data->new_catalog : GetWritable(data->old_catalog);

  // Set the previous revision hash in the new catalog to the old catalog
  // we are doing the whole migration as a new snapshot that does not change
  // any files, but just applies the necessary data schema migrations
  catalog->SetPreviousRevision(data->old_catalog->hash());
  catalog->IncrementRevision();
  catalog->UpdateLastModified();

  return true;
}


template<class DerivedT>
bool CommandMigrate::AbstractMigrationWorker<DerivedT>::
     CollectAndAggregateStatistics(PendingCatalog *data) const
{
  if (!collect_catalog_statistics_) {
    return true;
  }

  const catalog::Catalog *new_catalog =
    (data->HasNew()) ? data->new_catalog : data->old_catalog;
  const catalog::CatalogDatabase &writable = new_catalog->database();
  bool retval;

  // Find out the discrepancy between MAX(rowid) and COUNT(*)
  catalog::SqlCatalog wasted_inodes(writable,
    "SELECT COUNT(*), MAX(rowid) FROM catalog;");
  retval = wasted_inodes.FetchRow();
  if (!retval) {
    Error("Failed to count entries in catalog", wasted_inodes, data);
    return false;
  }
  const unsigned int entry_count = wasted_inodes.RetrieveInt64(0);
  const unsigned int max_row_id  = wasted_inodes.RetrieveInt64(1);

  // Save collected information into the central statistics aggregator
  data->statistics.root_path   = data->root_path();
  data->statistics.max_row_id  = max_row_id;
  data->statistics.entry_count = entry_count;

  return true;
}


template<class DerivedT>
bool CommandMigrate::AbstractMigrationWorker<DerivedT>::CleanupNestedCatalogs(
  PendingCatalog *data) const
{
  // All nested catalogs of PendingCatalog 'data' are fully processed and
  // accounted. It is safe to get rid of their data structures here!
  PendingCatalogList::const_iterator i    = data->nested_catalogs.begin();
  PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    delete *i;
  }

  data->nested_catalogs.clear();
  return true;
}


/**
 * Those values _must_ reflect the schema version in catalog_sql.h so that a
 * legacy catalog migration generates always the latest catalog revision.
 * This is a deliberately duplicated piece of information to ensure that always
 * both the catalog management and migration classes get updated.
 */
const float    CommandMigrate::MigrationWorker_20x::kSchema         = 2.5;
const unsigned CommandMigrate::MigrationWorker_20x::kSchemaRevision = 6;


template<class DerivedT>
catalog::WritableCatalog*
CommandMigrate::AbstractMigrationWorker<DerivedT>::GetWritable(
                                        const catalog::Catalog *catalog) const {
  return dynamic_cast<catalog::WritableCatalog*>(const_cast<catalog::Catalog*>(
    catalog));
}


//------------------------------------------------------------------------------


CommandMigrate::MigrationWorker_20x::MigrationWorker_20x(
  const worker_context *context)
  : AbstractMigrationWorker<MigrationWorker_20x>(context)
  , fix_nested_catalog_transitions_(context->fix_nested_catalog_transitions)
  , analyze_file_linkcounts_(context->analyze_file_linkcounts)
  , uid_(context->uid)
  , gid_(context->gid) { }


bool CommandMigrate::MigrationWorker_20x::RunMigration(PendingCatalog *data)
  const
{
  // double-check that we are generating compatible catalogs to the actual
  // catalog management classes
  assert(kSchema         == catalog::CatalogDatabase::kLatestSupportedSchema);
  assert(kSchemaRevision == catalog::CatalogDatabase::kLatestSchemaRevision);

  return CreateNewEmptyCatalog(data) &&
         CheckDatabaseSchemaCompatibility(data) &&
         AttachOldCatalogDatabase(data) &&
         StartDatabaseTransaction(data) &&
         MigrateFileMetadata(data) &&
         MigrateNestedCatalogMountPoints(data) &&
         FixNestedCatalogTransitionPoints(data) &&
         RemoveDanglingNestedMountpoints(data) &&
         GenerateCatalogStatistics(data) &&
         FindRootEntryInformation(data) &&
         CommitDatabaseTransaction(data) &&
         DetachOldCatalogDatabase(data);
}

bool CommandMigrate::MigrationWorker_20x::CreateNewEmptyCatalog(
  PendingCatalog *data) const
{
  const string root_path = data->root_path();

  // create a new catalog database schema
  const string clg_db_path =
    CreateTempPath(temporary_directory_ + "/catalog", 0666);
  if (clg_db_path.empty()) {
    Error("Failed to create temporary file for the new catalog database.");
    return false;
  }
  const bool volatile_content = false;

  {
    // TODO(rmeusel): Attach catalog should work with an open catalog database
    // as well, to remove this inefficiency
    UniquePtr<catalog::CatalogDatabase>
      new_clg_db(catalog::CatalogDatabase::Create(clg_db_path));
    if (!new_clg_db.IsValid() ||
        !new_clg_db->InsertInitialValues(root_path, volatile_content, "")) {
      Error("Failed to create database for new catalog");
      unlink(clg_db_path.c_str());
      return false;
    }
  }

  // Attach the just created nested catalog database
  catalog::WritableCatalog *writable_catalog =
    catalog::WritableCatalog::AttachFreely(root_path, clg_db_path,
                                           shash::Any(shash::kSha1));
  if (writable_catalog == NULL) {
    Error("Failed to open database for new catalog");
    unlink(clg_db_path.c_str());
    return false;
  }

  data->new_catalog = writable_catalog;
  return true;
}


bool CommandMigrate::MigrationWorker_20x::CheckDatabaseSchemaCompatibility(
  PendingCatalog *data) const
{
  const catalog::CatalogDatabase &old_catalog = data->old_catalog->database();
  const catalog::CatalogDatabase &new_catalog = data->new_catalog->database();

  if ((new_catalog.schema_version() <
         catalog::CatalogDatabase::kLatestSupportedSchema -
         catalog::CatalogDatabase::kSchemaEpsilon
       ||
       new_catalog.schema_version() >
         catalog::CatalogDatabase::kLatestSupportedSchema +
         catalog::CatalogDatabase::kSchemaEpsilon)
       ||
       (old_catalog.schema_version() > 2.1 +
         catalog::CatalogDatabase::kSchemaEpsilon))
  {
    Error("Failed to meet database requirements for migration.", data);
    return false;
  }
  return true;
}


bool CommandMigrate::MigrationWorker_20x::AttachOldCatalogDatabase(
  PendingCatalog *data) const
{
  const catalog::CatalogDatabase &old_catalog = data->old_catalog->database();
  const catalog::CatalogDatabase &new_catalog = data->new_catalog->database();

  catalog::SqlCatalog sql_attach_new(new_catalog,
    "ATTACH '" + old_catalog.filename() + "' AS old;");
  bool retval = sql_attach_new.Execute();

  // remove the hardlink to the old database file (temporary file), it will not
  // be needed anymore... data will get deleted when the database is closed
  unlink(data->old_catalog->database().filename().c_str());

  if (!retval) {
    Error("Failed to attach database of old catalog", sql_attach_new, data);
    return false;
  }
  return true;
}


bool CommandMigrate::MigrationWorker_20x::StartDatabaseTransaction(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  data->new_catalog->Transaction();
  return true;
}


bool CommandMigrate::MigrationWorker_20x::MigrateFileMetadata(
  PendingCatalog *data) const
{
  assert(!data->new_catalog->IsDirty());
  assert(data->HasNew());
  bool retval;
  const catalog::CatalogDatabase &writable = data->new_catalog->database();

  // Hardlinks scratch space.
  // This temporary table is used for the hardlink analysis results.
  // The old catalog format did not have a direct notion of hardlinks and their
  // linkcounts,  but this information can be partly retrieved from the under-
  // lying file system semantics.
  //
  //   Hardlinks:
  //     groupid   : this group id can be used for the new catalog schema
  //     inode     : the inodes that were part of a hardlink group before
  //     linkcount : the linkcount for hardlink group id members
  catalog::SqlCatalog sql_create_hardlinks_table(writable,
    "CREATE TEMPORARY TABLE hardlinks "
    "  (  hardlink_group_id  INTEGER PRIMARY KEY AUTOINCREMENT, "
    "     inode              INTEGER, "
    "     linkcount          INTEGER, "
    "     CONSTRAINT unique_inode UNIQUE (inode)  );");
  retval = sql_create_hardlinks_table.Execute();
  if (!retval) {
    Error("Failed to create temporary hardlink analysis table",
          sql_create_hardlinks_table, data);
    return false;
  }

  // Directory Linkcount scratch space.
  // Directory linkcounts can be obtained from the directory hierarchy reflected
  // in the old style catalogs. The new catalog schema asks for this specific
  // linkcount. Directory linkcount analysis results will be put into this
  // temporary table
  catalog::SqlCatalog sql_create_linkcounts_table(writable,
    "CREATE TEMPORARY TABLE dir_linkcounts "
    "  (  inode      INTEGER PRIMARY KEY, "
    "     linkcount  INTEGER  );");
  retval = sql_create_linkcounts_table.Execute();
  if (!retval) {
    Error("Failed to create tmeporary directory linkcount analysis table",
          sql_create_linkcounts_table, data);
  }

  // It is possible to skip this step.
  // In that case all hardlink inodes with a (potential) linkcount > 1 will get
  // degraded to files containing the same content
  if (analyze_file_linkcounts_) {
    retval = AnalyzeFileLinkcounts(data);
    if (!retval) {
      return false;
    }
  }

  // Analyze the linkcounts of directories
  //   - each directory has a linkcount of at least 2 (empty directory)
  //     (link in parent directory and self reference (cd .) )
  //   - for each child directory, the parent's link count is incremented by 1
  //     (parent reference in child (cd ..) )
  //
  // Note: nested catalog mountpoints will be miscalculated here, since we can't
  //       check the number of containing directories. They are defined in a the
  //       linked nested catalog and need to be added later on.
  //       (see: MigrateNestedCatalogMountPoints() for details)
  catalog::SqlCatalog sql_dir_linkcounts(writable,
    "INSERT INTO dir_linkcounts "
    "  SELECT c1.inode as inode, "
    "         SUM(IFNULL(MIN(c2.inode,1),0)) + 2 as linkcount "
    "  FROM old.catalog as c1 "
    "  LEFT JOIN old.catalog as c2 "
    "    ON c2.parent_1 = c1.md5path_1 AND "
    "       c2.parent_2 = c1.md5path_2 AND "
    "       c2.flags & :flag_dir_1 "
    "  WHERE c1.flags & :flag_dir_2 "
    "  GROUP BY c1.inode;");
  retval =
    sql_dir_linkcounts.BindInt64(1, catalog::SqlDirent::kFlagDir) &&
    sql_dir_linkcounts.BindInt64(2, catalog::SqlDirent::kFlagDir) &&
    sql_dir_linkcounts.Execute();
  if (!retval) {
    Error("Failed to analyze directory specific linkcounts",
          sql_dir_linkcounts, data);
    if (sql_dir_linkcounts.GetLastError() == SQLITE_CONSTRAINT) {
      Error("Obviously your catalogs are corrupted, since we found a directory"
            "inode that is a file inode at the same time!");
    }
    return false;
  }

  // Copy the old file meta information into the new catalog schema
  //   here we also add the previously analyzed hardlink/linkcount information
  //   from both temporary tables "hardlinks" and "dir_linkcounts".
  //
  // Note: nested catalog mountpoints still need to be treated separately
  //       (see MigrateNestedCatalogMountPoints() for details)
  catalog::SqlCatalog migrate_file_meta_data(writable,
    "INSERT INTO catalog "
    "  SELECT md5path_1, md5path_2, "
    "         parent_1, parent_2, "
    "         IFNULL(hardlink_group_id, 0) << 32 | "
    "         COALESCE(hardlinks.linkcount, dir_linkcounts.linkcount, 1) "
    "           AS hardlinks, "
    "         hash, size, mode, mtime, "
    "         flags, name, symlink, "
    "         :uid, "
    "         :gid, "
    "         NULL "  // set empty xattr BLOB (default)
    "  FROM old.catalog "
    "  LEFT JOIN hardlinks "
    "    ON catalog.inode = hardlinks.inode "
    "  LEFT JOIN dir_linkcounts "
    "    ON catalog.inode = dir_linkcounts.inode;");
  retval = migrate_file_meta_data.BindInt64(1, uid_) &&
           migrate_file_meta_data.BindInt64(2, gid_) &&
           migrate_file_meta_data.Execute();
  if (!retval) {
    Error("Failed to migrate the file system meta data",
          migrate_file_meta_data, data);
    return false;
  }

  // If we deal with a nested catalog, we need to add a .cvmfscatalog entry
  // since it was not present in the old repository specification but is needed
  // now!
  if (!data->IsRoot()) {
    const catalog::DirectoryEntry &nested_marker =
      CommandMigrate::GetNestedCatalogMarkerDirent();
    catalog::SqlDirentInsert insert_nested_marker(writable);
    const std::string   root_path   = data->root_path();
    const std::string   file_path   = root_path +
                                      "/" + nested_marker.name().ToString();
    const shash::Md5    &path_hash   = shash::Md5(file_path.data(),
                                                  file_path.size());
    const shash::Md5    &parent_hash = shash::Md5(root_path.data(),
                                                  root_path.size());
    retval = insert_nested_marker.BindPathHash(path_hash)         &&
             insert_nested_marker.BindParentPathHash(parent_hash) &&
             insert_nested_marker.BindDirent(nested_marker)       &&
             insert_nested_marker.BindXattrEmpty()                &&
             insert_nested_marker.Execute();
    if (!retval) {
      Error("Failed to insert nested catalog marker into new nested catalog.",
            insert_nested_marker, data);
      return false;
    }
  }

  // Copy (and update) the properties fields
  //
  // Note: The 'schema' is explicitly not copied to the new catalog.
  //       Each catalog contains a revision, which is also copied here and that
  //       is later updated by calling catalog->IncrementRevision()
  catalog::SqlCatalog copy_properties(writable,
    "INSERT OR REPLACE INTO properties "
    "  SELECT key, value "
    "  FROM old.properties "
    "  WHERE key != 'schema';");
  retval = copy_properties.Execute();
  if (!retval) {
    Error("Failed to migrate the properties table.", copy_properties, data);
    return false;
  }

  return true;
}


bool CommandMigrate::MigrationWorker_20x::AnalyzeFileLinkcounts(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  const catalog::CatalogDatabase &writable = data->new_catalog->database();
  bool retval;

  // Analyze the hardlink relationships in the old catalog
  //   inodes used to be assigned at publishing time, implicitly constituating
  //   those relationships. We now need them explicitly in the file catalogs
  // This looks for directory entries with matching inodes but differing path-
  // hashes and saves the results in a temporary table called 'hl_scratch'
  //
  // Note: We only support hardlink groups that reside in the same directory!
  //       Therefore we first need to figure out hardlink candidates (which
  //       might still contain hardlink groups spanning more than one directory)
  //       In a second step these candidates will be analyzed to kick out un-
  //       supported hardlink groups.
  //       Unsupported hardlink groups will be be treated as normal files with
  //       the same content
  catalog::SqlCatalog sql_create_hardlinks_scratch_table(writable,
    "CREATE TEMPORARY TABLE hl_scratch AS "
    "  SELECT c1.inode AS inode, c1.md5path_1, c1.md5path_2, "
    "         c1.parent_1 as c1p1, c1.parent_2 as c1p2, "
    "         c2.parent_1 as c2p1, c2.parent_2 as c2p2 "
    "  FROM old.catalog AS c1 "
    "  INNER JOIN old.catalog AS c2 "
    "  ON c1.inode == c2.inode AND "
    "     (c1.md5path_1 != c2.md5path_1 OR "
    "      c1.md5path_2 != c2.md5path_2);");
  retval = sql_create_hardlinks_scratch_table.Execute();
  if (!retval) {
    Error("Failed to create temporary scratch table for hardlink analysis",
          sql_create_hardlinks_scratch_table, data);
    return false;
  }

  // Figures out which hardlink candidates are supported by CVMFS and can be
  // transferred into the new catalog as so called hardlink groups. Unsupported
  // hardlinks need to be discarded and treated as normal files containing the
  // exact same data
  catalog::SqlCatalog fill_linkcount_table_for_files(writable,
    "INSERT INTO hardlinks (inode, linkcount)"
    "  SELECT inode, count(*) as linkcount "
    "  FROM ( "
         // recombine supported hardlink inodes with their actual manifested
         // hard-links in the catalog.
         // Note: for each directory entry pointing to the same supported
         // hardlink inode we have a distinct MD5 path hash
    "    SELECT DISTINCT hl.inode, hl.md5path_1, hl.md5path_2 "
    "    FROM ( "
           // sort out supported hardlink inodes from unsupported ones by
           // locality
           // Note: see the next comment for the nested SELECT
    "      SELECT inode "
    "      FROM ( "
    "        SELECT inode, count(*) AS cnt "
    "        FROM ( "
               // go through the potential hardlinks and collect location infor-
               // mation about them.
               // Note: we only support hardlinks that all reside in the same
               //       directory, thus having the same parent (c1p* == c2p*)
               //   --> For supported hardlink candidates the SELECT DISTINCT
               // will produce only a single row, whereas others produce more
    "          SELECT DISTINCT inode,c1p1,c1p1,c2p1,c2p2 "
    "          FROM hl_scratch AS hl "
    "        ) "
    "        GROUP BY inode "
    "      ) "
    "      WHERE cnt = 1 "
    "    ) AS supported_hardlinks "
    "    LEFT JOIN hl_scratch AS hl "
    "    ON supported_hardlinks.inode = hl.inode "
    "  ) "
    "  GROUP BY inode;");
  retval = fill_linkcount_table_for_files.Execute();
  if (!retval) {
    Error("Failed to analyze hardlink relationships for files.",
          fill_linkcount_table_for_files, data);
    return false;
  }

  // The file linkcount and hardlink analysis is finished and the scratch table
  // can be deleted...
  catalog::SqlCatalog drop_hardlink_scratch_space(writable,
                                                  "DROP TABLE hl_scratch;");
  retval = drop_hardlink_scratch_space.Execute();
  if (!retval) {
    Error("Failed to remove file linkcount analysis scratch table",
          drop_hardlink_scratch_space, data);
    return false;
  }

  // Do some statistics if asked for...
  if (collect_catalog_statistics_) {
    catalog::SqlCatalog count_hardlinks(writable,
      "SELECT count(*), sum(linkcount) FROM hardlinks;");
    retval = count_hardlinks.FetchRow();
    if (!retval) {
      Error("Failed to count the generated file hardlinks for statistics",
            count_hardlinks, data);
      return false;
    }

    data->statistics.hardlink_group_count  += count_hardlinks.RetrieveInt64(0);
    data->statistics.aggregated_linkcounts += count_hardlinks.RetrieveInt64(1);
  }

  return true;
}


bool CommandMigrate::MigrationWorker_20x::MigrateNestedCatalogMountPoints(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  const catalog::CatalogDatabase &writable = data->new_catalog->database();
  bool retval;

  // preparing the SQL statement for nested catalog mountpoint update
  catalog::SqlCatalog update_mntpnt_linkcount(writable,
    "UPDATE catalog "
    "SET hardlinks = :linkcount "
    "WHERE md5path_1 = :md5_1 AND md5path_2 = :md5_2;");

  // update all nested catalog mountpoints
  // (Note: we might need to wait for the nested catalog to be processed)
}


/**
 * Queries a single catalog and looks for DirectoryEntrys that have direct
 * children in the same catalog but are marked as 'nested catalog mountpoints'.
 * This is an inconsistent situation, since a mountpoint is supposed to be empty
 * and it's children are stored in the corresponding referenced nested catalog.
 *
 * Note: the user code needs to check if there is a corresponding nested catalog
 *       reference for the found dangling mountpoints. If so, we also have a
 *       bogus state, but it is not reliably fixable automatically. The child-
 *       DirectoryEntrys would be masked by the mounting nested catalog but it
 *       is not clear if we can simply delete them or if this would destroy data.
 */
class SqlLookupDanglingMountpoints : public catalog::SqlLookup {
 public:
  explicit SqlLookupDanglingMountpoints(
    const catalog::CatalogDatabase &database)
  {
    const string statement =
      "SELECT DISTINCT " +
      GetFieldsToSelect(database.schema_version(), database.schema_revision()) +
      " FROM catalog "
      "JOIN catalog AS c2 "
      "ON catalog.md5path_1 = c2.parent_1 AND catalog.md5path_2 = c2.parent_2 "
      "WHERE catalog.flags & "
      + StringifyInt(catalog::SqlDirent::kFlagDirNestedMountpoint) + ";";
    const bool init_successful = Init(database.sqlite_db(), statement);
    assert(init_successful);
  }
};

bool CommandMigrate::MigrationWorker_20x::RemoveDanglingNestedMountpoints(
  PendingCatalog *data) const
{
    const string &root_path = nested_catalog->root_path();

    // update the nested catalog mountpoint directory entry with the correct
    // linkcount that was determined while processing the nested catalog
    const shash::Md5 mountpoint_hash = shash::Md5(root_path.data(),
                                                  root_path.size());
    retval =
      update_mntpnt_linkcount.BindInt64(1, root_entry.linkcount()) &&
      update_mntpnt_linkcount.BindMd5(2, 3, mountpoint_hash)       &&
      update_mntpnt_linkcount.Execute();
    if (!retval) {
      Error("Failed to update linkcount of nested catalog mountpoint",
            update_mntpnt_linkcount, data);
      return false;
    }
    update_mntpnt_linkcount.Reset();
  }


  // Retrieve nested catalog mountpoints that have child entries directly inside
  // the current catalog (which is a malformed state)
  SqlLookupDanglingMountpoints sql_dangling_mountpoints(writable);
  catalog::SqlDirentUpdate save_updated_mountpoint(writable);

  // go through the list of dangling nested catalog mountpoints and fix them
  assert(data->HasNew());
  if (!fix_nested_catalog_transitions_) {
    // Fixing transition point mismatches is not enabled...
    return true;
  }

  typedef catalog::DirectoryEntry::Difference Difference;

  const catalog::CatalogDatabase &writable = data->new_catalog->database();
  bool retval;

  catalog::SqlLookupPathHash lookup_mountpoint(writable);
  catalog::SqlDirentUpdate   update_directory_entry(writable);

  // Unbox the nested catalogs (possibly waiting for migration of them first)
  PendingCatalogList::const_iterator i    = data->nested_catalogs.begin();
  PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    // Collect information about the nested catalog
    PendingCatalog *nested_catalog = *i;
    const catalog::DirectoryEntry nested_root_entry =
      nested_catalog->root_entry.Get();
    const string &nested_root_path = nested_catalog->root_path();
    const shash::Md5 mountpoint_path_hash =
      shash::Md5(nested_root_path.data(), nested_root_path.size());

    // Retrieve the nested catalog mountpoint from the current catalog
    retval = lookup_mountpoint.BindPathHash(mountpoint_path_hash) &&
             lookup_mountpoint.FetchRow();
    if (!retval) {
      Error("Failed to fetch nested catalog mountpoint to check for compatible"
            "transition points", lookup_mountpoint, data);
      return false;
    }

    catalog::DirectoryEntry mountpoint_entry =
      lookup_mountpoint.GetDirent(data->new_catalog);
    lookup_mountpoint.Reset();

    // Compare nested catalog mountpoint and nested catalog root entries
    catalog::DirectoryEntry::Differences diffs =
      mountpoint_entry.CompareTo(nested_root_entry);

    // We MUST deal with two directory entries that are a pair of nested cata-
    // log mountpoint and root entry! Thus we expect their transition flags to
    // differ and their name to be the same.
    assert(diffs & Difference::kNestedCatalogTransitionFlags);
    assert((diffs & Difference::kName) == 0);

    // Check if there are other differences except the nested catalog transition
    // flags and fix them...
    if ((diffs ^ Difference::kNestedCatalogTransitionFlags) != 0) {
      // If we found differences, we still assume a couple of directory entry
      // fields to be the same, otherwise some severe stuff would be wrong...
      if ((diffs & Difference::kChecksum)        ||
          (diffs & Difference::kLinkcount)       ||
          (diffs & Difference::kSymlink)         ||
          (diffs & Difference::kChunkedFileFlag)    )
      {
        Error("Found an irreparable mismatch in a nested catalog transition "
              "point at '" + nested_root_path + "'\nAborting...\n");
      }

      // Copy the properties from the nested catalog root entry into the mount-
      // point entry to bring them in sync again
      CommandMigrate::FixNestedCatalogTransitionPoint(
        nested_root_entry, &mountpoint_entry);

      // save the nested catalog mountpoint entry into the catalog
      retval = update_directory_entry.BindPathHash(mountpoint_path_hash) &&
               update_directory_entry.BindDirent(mountpoint_entry)       &&
               update_directory_entry.Execute();
      if (!retval) {
        Error("Failed to save resynchronized nested catalog mountpoint into "
              "catalog database", update_directory_entry, data);
        return false;
      }
      update_directory_entry.Reset();

      // Fixing of this mountpoint went well... inform the user that this minor
      // issue occured
      LogCvmfs(kLogCatalog, kLogStdout,
               "NOTE: fixed incompatible nested catalog transition point at: "
               "'%s' ", nested_root_path.c_str());
    }
  }

  return true;
}


void CommandMigrate::FixNestedCatalogTransitionPoint(
  const catalog::DirectoryEntry &nested_root,
  catalog::DirectoryEntry *mountpoint
) {
  // Replace some file system parameters in the mountpoint to resync it with
  // the nested root of the corresponding nested catalog
  //
  // Note: this method relies on CommandMigrate being a friend of DirectoryEntry
  mountpoint->mode_  = nested_root.mode_;
  mountpoint->uid_   = nested_root.uid_;
  mountpoint->gid_   = nested_root.gid_;
  mountpoint->size_  = nested_root.size_;
  mountpoint->mtime_ = nested_root.mtime_;
}


bool CommandMigrate::MigrationWorker_20x::RemoveDanglingNestedMountpoints(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  const catalog::CatalogDatabase &writable = data->new_catalog->database();
  bool retval = false;

  // build a set of registered nested catalog path hashes
  typedef catalog::Catalog::NestedCatalogList NestedCatalogList;
  typedef std::map<shash::Md5, catalog::Catalog::NestedCatalog>
    NestedCatalogMap;
  const NestedCatalogList& nested_clgs =
    data->old_catalog->ListNestedCatalogs();
  NestedCatalogList::const_iterator i = nested_clgs.begin();
  const NestedCatalogList::const_iterator iend = nested_clgs.end();
  NestedCatalogMap nested_catalog_path_hashes;
  for (; i != iend; ++i) {
    const PathString &path = i->mountpoint;
    const shash::Md5  hash(path.GetChars(), path.GetLength());
    nested_catalog_path_hashes[hash] = *i;
  }

  // Retrieve nested catalog mountpoints that have child entries directly inside
  // the current catalog (which is a malformed state)
  catalog::SqlLookupDanglingMountpoints sql_dangling_mountpoints(writable);
  catalog::SqlDirentUpdate save_updated_mountpoint(writable);

  std::vector<catalog::DirectoryEntry> todo_dirent;
  std::vector<shash::Md5> todo_hash;

  // go through the list of dangling nested catalog mountpoints and fix them
  // where needed (check if there is no nested catalog registered for them)
  while (sql_dangling_mountpoints.FetchRow()) {
    catalog::DirectoryEntry dangling_mountpoint =
      sql_dangling_mountpoints.GetDirent(data->new_catalog);
    const shash::Md5 path_hash = sql_dangling_mountpoints.GetPathHash();
    assert(dangling_mountpoint.IsNestedCatalogMountpoint());

    // check if the nested catalog mountpoint is registered in the nested cata-
    // log list of the currently migrated catalog
    const NestedCatalogMap::const_iterator nested_catalog =
                                     nested_catalog_path_hashes.find(path_hash);
    if (nested_catalog != nested_catalog_path_hashes.end()) {
      LogCvmfs(kLogCatalog, kLogStderr,
               "WARNING: found a non-empty nested catalog mountpoint under "
               "'%s'", nested_catalog->second.mountpoint.c_str());
      continue;
    }

    // the mountpoint was confirmed to be dangling and needs to be removed
    dangling_mountpoint.set_is_nested_catalog_mountpoint(false);
    todo_dirent.push_back(dangling_mountpoint);
    todo_hash.push_back(path_hash);
  }

  for (unsigned i = 0; i < todo_dirent.size(); ++i) {
    retval = save_updated_mountpoint.BindPathHash(todo_hash[i])  &&
             save_updated_mountpoint.BindDirent(todo_dirent[i])  &&
             save_updated_mountpoint.Execute()                   &&
             save_updated_mountpoint.Reset();
    if (!retval) {
      Error("Failed to remove dangling nested catalog mountpoint entry in "
            "catalog", save_updated_mountpoint, data);
      return false;
    }

    // tell the user that this intervention has been taken place
    LogCvmfs(kLogCatalog, kLogStdout, "NOTE: fixed dangling nested catalog "
                                      "mountpoint entry called: '%s' ",
                                      todo_dirent[i].name().c_str());
  }

  return true;
}


const catalog::DirectoryEntry& CommandMigrate::GetNestedCatalogMarkerDirent() {
  // This is pre-initialized singleton... it MUST be already there...
  assert(nested_catalog_marker_.name_.ToString() == ".cvmfscatalog");
  return nested_catalog_marker_;
}

bool CommandMigrate::GenerateNestedCatalogMarkerChunk() {
  // Create an empty nested catalog marker file
  nested_catalog_marker_tmp_path_ =
      CreateTempPath(temporary_directory_ + "/.cvmfscatalog", 0644);
  if (nested_catalog_marker_tmp_path_.empty()) {
    Error("Failed to create temp file for nested catalog marker dummy.");
    return false;
  }

  // Process and upload it to the backend storage
  IngestionSource *source =
      new FileIngestionSource(nested_catalog_marker_tmp_path_);
  spooler_->Process(source);
  return true;
}

void CommandMigrate::CreateNestedCatalogMarkerDirent(
  const shash::Any &content_hash)
{
  // Generate it only once
  assert(nested_catalog_marker_.name_.ToString() != ".cvmfscatalog");

  // Fill the DirectoryEntry structure will all needed information
  nested_catalog_marker_.name_.Assign(".cvmfscatalog", strlen(".cvmfscatalog"));
  nested_catalog_marker_.mode_      = 33188;
  nested_catalog_marker_.uid_       = uid_;
  nested_catalog_marker_.gid_       = gid_;
  nested_catalog_marker_.size_      = 0;
  nested_catalog_marker_.mtime_     = time(NULL);
  nested_catalog_marker_.linkcount_ = 1;
  nested_catalog_marker_.checksum_  = content_hash;
}


bool CommandMigrate::MigrationWorker_20x::GenerateCatalogStatistics(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  bool retval = false;
  const catalog::CatalogDatabase &writable = data->new_catalog->database();

  // Aggregated the statistics counters of all nested catalogs
  // Note: we might need to wait until nested catalogs are sucessfully processed
  catalog::DeltaCounters stats_counters;
  PendingCatalogList::const_iterator i    = data->nested_catalogs.begin();
  PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    const PendingCatalog *nested_catalog = *i;
    const catalog::DeltaCounters &s = nested_catalog->nested_statistics.Get();
    s.PopulateToParent(&stats_counters);
  }

  // Count various directory entry types in the catalog to fill up the catalog
  // statistics counters introduced in the current catalog schema
  catalog::SqlCatalog count_regular_files(writable,
    "SELECT count(*) FROM catalog "
    "                WHERE  flags & :flag_file "
    "                       AND NOT flags & :flag_link;");
  catalog::SqlCatalog count_symlinks(writable,
    "SELECT count(*) FROM catalog WHERE flags & :flag_link;");
  catalog::SqlCatalog count_directories(writable,
    "SELECT count(*) FROM catalog WHERE flags & :flag_dir;");
  catalog::SqlCatalog aggregate_file_size(writable,
    "SELECT sum(size) FROM catalog WHERE  flags & :flag_file "
    "                                     AND NOT flags & :flag_link");

  // Run the actual counting queries
  retval =
    count_regular_files.BindInt64(1, catalog::SqlDirent::kFlagFile) &&
    count_regular_files.BindInt64(2, catalog::SqlDirent::kFlagLink) &&
    count_regular_files.FetchRow();
  if (!retval) {
    Error("Failed to count regular files.", count_regular_files, data);
    return false;
  }
  retval =
    count_symlinks.BindInt64(1, catalog::SqlDirent::kFlagLink) &&
    count_symlinks.FetchRow();
  if (!retval) {
    Error("Failed to count symlinks.", count_symlinks, data);
    return false;
  }
  retval =
    count_directories.BindInt64(1, catalog::SqlDirent::kFlagDir) &&
    count_directories.FetchRow();
  if (!retval) {
    Error("Failed to count directories.", count_directories, data);
    return false;
  }
  retval =
    aggregate_file_size.BindInt64(1, catalog::SqlDirent::kFlagFile) &&
    aggregate_file_size.BindInt64(2, catalog::SqlDirent::kFlagLink) &&
    aggregate_file_size.FetchRow();
  if (!retval) {
    Error("Failed to aggregate the file sizes.", aggregate_file_size, data);
    return false;
  }

  // Insert the counted statistics into the DeltaCounters data structure
  stats_counters.self.regular_files    = count_regular_files.RetrieveInt64(0);
  stats_counters.self.symlinks         = count_symlinks.RetrieveInt64(0);
  stats_counters.self.directories      = count_directories.RetrieveInt64(0);
  stats_counters.self.nested_catalogs  = data->nested_catalogs.size();
  stats_counters.self.file_size        = aggregate_file_size.RetrieveInt64(0);

  // Write back the generated statistics counters into the catalog database
  stats_counters.WriteToDatabase(writable);

  // Push the generated statistics counters up to the parent catalog
  data->nested_statistics.Set(stats_counters);

  return true;
}


bool CommandMigrate::MigrationWorker_20x::FindRootEntryInformation(
  PendingCatalog *data) const
{
  const catalog::CatalogDatabase &writable = data->new_catalog->database();
  bool retval;

  std::string root_path = data->root_path();
  shash::Md5 root_path_hash = shash::Md5(root_path.data(), root_path.size());

  catalog::SqlLookupPathHash lookup_root_entry(writable);
  retval = lookup_root_entry.BindPathHash(root_path_hash) &&
           lookup_root_entry.FetchRow();
  if (!retval) {
    Error("Failed to retrieve root directory entry of migrated catalog",
          lookup_root_entry, data);
    return false;
  }

  catalog::DirectoryEntry entry =
    lookup_root_entry.GetDirent(data->new_catalog);
  if (entry.linkcount() < 2 || entry.hardlink_group() > 0) {
    Error("Retrieved linkcount of catalog root entry is not sane.", data);
    return false;
  }

  data->root_entry.Set(entry);
  return true;
}


bool CommandMigrate::MigrationWorker_20x::CommitDatabaseTransaction(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  data->new_catalog->Commit();
  return true;
}


bool CommandMigrate::MigrationWorker_20x::DetachOldCatalogDatabase(
  PendingCatalog *data) const
{
  assert(data->HasNew());
  const catalog::CatalogDatabase &writable = data->new_catalog->database();
  catalog::SqlCatalog detach_old_catalog(writable, "DETACH old;");
  const bool retval = detach_old_catalog.Execute();
  if (!retval) {
    Error("Failed to detach old catalog database.", detach_old_catalog, data);
    return false;
  }
  return true;
}


//------------------------------------------------------------------------------


CommandMigrate::MigrationWorker_217::MigrationWorker_217(
  const worker_context *context)
  : AbstractMigrationWorker<MigrationWorker_217>(context)
{ }


bool CommandMigrate::MigrationWorker_217::RunMigration(PendingCatalog *data)
  const
{
  return CheckDatabaseSchemaCompatibility(data) &&
         StartDatabaseTransaction(data) &&
         GenerateNewStatisticsCounters(data) &&
         UpdateCatalogSchema(data) &&
         CommitDatabaseTransaction(data);
}


bool CommandMigrate::MigrationWorker_217::CheckDatabaseSchemaCompatibility(
  PendingCatalog *data) const
{
  assert(!data->HasNew());
  const catalog::CatalogDatabase &old_catalog = data->old_catalog->database();

  if ((old_catalog.schema_version() < 2.4 -
       catalog::CatalogDatabase::kSchemaEpsilon)
      ||
      (old_catalog.schema_version() > 2.4 +
       catalog::CatalogDatabase::kSchemaEpsilon))
  {
    Error("Given Catalog is not Schema 2.4.", data);
    return false;
  }

  return true;
}


bool CommandMigrate::MigrationWorker_217::StartDatabaseTransaction(
  PendingCatalog *data) const
{
  assert(!data->HasNew());
  GetWritable(data->old_catalog)->Transaction();
  return true;
}


bool CommandMigrate::MigrationWorker_217::GenerateNewStatisticsCounters
                                                  (PendingCatalog *data) const {
  assert(!data->HasNew());
  bool retval = false;
  const catalog::CatalogDatabase &writable =
    GetWritable(data->old_catalog)->database();

  // Aggregated the statistics counters of all nested catalogs
  // Note: we might need to wait until nested catalogs are sucessfully processed
  catalog::DeltaCounters stats_counters;
  PendingCatalogList::const_iterator i = data->nested_catalogs.begin();
  PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    const PendingCatalog *nested_catalog = *i;
    const catalog::DeltaCounters &s = nested_catalog->nested_statistics.Get();
    s.PopulateToParent(&stats_counters);
  }

  // Count various directory entry types in the catalog to fill up the catalog
  // statistics counters introduced in the current catalog schema
  catalog::SqlCatalog count_chunked_files(writable,
    "SELECT count(*), sum(size) FROM catalog "
    "                WHERE flags & :flag_chunked_file;");
  catalog::SqlCatalog count_file_chunks(writable,
    "SELECT count(*) FROM chunks;");
  catalog::SqlCatalog aggregate_file_size(writable,
    "SELECT sum(size) FROM catalog WHERE  flags & :flag_file "
    "                                     AND NOT flags & :flag_link;");

  // Run the actual counting queries
  retval =
    count_chunked_files.BindInt64(1, catalog::SqlDirent::kFlagFileChunk) &&
    count_chunked_files.FetchRow();
  if (!retval) {
    Error("Failed to count chunked files.", count_chunked_files, data);
    return false;
  }
  retval = count_file_chunks.FetchRow();
  if (!retval) {
    Error("Failed to count file chunks", count_file_chunks, data);
    return false;
  }
  retval =
    aggregate_file_size.BindInt64(1, catalog::SqlDirent::kFlagFile) &&
    aggregate_file_size.BindInt64(2, catalog::SqlDirent::kFlagLink) &&
    aggregate_file_size.FetchRow();
  if (!retval) {
    Error("Failed to aggregate the file sizes.", aggregate_file_size, data);
    return false;
  }

  // Insert the counted statistics into the DeltaCounters data structure
  stats_counters.self.chunked_files     = count_chunked_files.RetrieveInt64(0);
  stats_counters.self.chunked_file_size = count_chunked_files.RetrieveInt64(1);
  stats_counters.self.file_chunks       = count_file_chunks.RetrieveInt64(0);
  stats_counters.self.file_size         = aggregate_file_size.RetrieveInt64(0);

  // Write back the generated statistics counters into the catalog database
  catalog::Counters counters;
  retval = counters.ReadFromDatabase(writable, catalog::LegacyMode::kLegacy);
  if (!retval) {
    Error("Failed to read old catalog statistics counters", data);
    return false;
  }
  counters.ApplyDelta(stats_counters);
  retval = counters.InsertIntoDatabase(writable);
  if (!retval) {
    Error("Failed to write new statistics counters to database", data);
    return false;
  }

  // Push the generated statistics counters up to the parent catalog
  data->nested_statistics.Set(stats_counters);

  return true;
}


bool CommandMigrate::MigrationWorker_217::UpdateCatalogSchema
                                                  (PendingCatalog *data) const {
  assert(!data->HasNew());
  const catalog::CatalogDatabase &writable =
    GetWritable(data->old_catalog)->database();
  catalog::SqlCatalog update_schema_version(writable,
    "UPDATE properties SET value = :schema_version WHERE key = 'schema';");

  const bool retval =
    update_schema_version.BindDouble(1, 2.5) &&
    update_schema_version.Execute();
  if (!retval) {
    Error("Failed to update catalog schema version",
          update_schema_version,
          data);
    return false;
  }

  return true;
}


bool CommandMigrate::MigrationWorker_217::CommitDatabaseTransaction
                                                  (PendingCatalog *data) const {
  assert(!data->HasNew());
  GetWritable(data->old_catalog)->Commit();
  return true;
}


//------------------------------------------------------------------------------


CommandMigrate::ChownMigrationWorker::ChownMigrationWorker(
                                                const worker_context *context)
  : AbstractMigrationWorker<ChownMigrationWorker>(context)
  , uid_map_statement_(GenerateMappingStatement(context->uid_map, "uid"))
  , gid_map_statement_(GenerateMappingStatement(context->gid_map, "gid"))
{}

bool CommandMigrate::ChownMigrationWorker::RunMigration(
                                                   PendingCatalog *data) const {
  return ApplyPersonaMappings(data);
}


bool CommandMigrate::ChownMigrationWorker::ApplyPersonaMappings(
                                                   PendingCatalog *data) const {
  assert(data->old_catalog != NULL);
  assert(data->new_catalog == NULL);

  if (data->old_catalog->mountpoint() ==
      PathString("/" + string(catalog::VirtualCatalog::kVirtualPath)))
  {
    // skipping virtual catalog
    return true;
  }

  const catalog::CatalogDatabase &db =
                                     GetWritable(data->old_catalog)->database();

  if (!db.BeginTransaction()) {
    return false;
  }

  catalog::SqlCatalog uid_sql(db, uid_map_statement_);
  if (!uid_sql.Execute()) {
    Error("Failed to update UIDs", uid_sql, data);
    return false;
  }

  catalog::SqlCatalog gid_sql(db, gid_map_statement_);
  if (!gid_sql.Execute()) {
    Error("Failed to update GIDs", gid_sql, data);
    return false;
  }

  return db.CommitTransaction();
}


template <class MapT>
std::string CommandMigrate::ChownMigrationWorker::GenerateMappingStatement(
                                             const MapT         &map,
                                             const std::string  &column) const {
  assert(map.RuleCount() > 0 || map.HasDefault());

  std::string stmt = "UPDATE OR ABORT catalog SET " + column + " = ";

  if (map.RuleCount() == 0) {
    // map everything to the same value (just a simple UPDATE clause)
    stmt += StringifyInt(map.GetDefault());
  } else {
    // apply multiple ID mappings (UPDATE clause with CASE statement)
    stmt += "CASE " + column + " ";
    typedef typename MapT::map_type::const_iterator map_iterator;
          map_iterator i    = map.GetRuleMap().begin();
    const map_iterator iend = map.GetRuleMap().end();
    for (; i != iend; ++i) {
      stmt += "WHEN " + StringifyInt(i->first) +
             " THEN " + StringifyInt(i->second) + " ";
    }

    // add a default (if provided) or leave unchanged if no mapping fits
    stmt += (map.HasDefault())
                ? "ELSE " + StringifyInt(map.GetDefault()) + " "
                : "ELSE " + column + " ";
    stmt += "END";
  }

  stmt += ";";
  return stmt;
}


//------------------------------------------------------------------------------


bool CommandMigrate::HardlinkRemovalMigrationWorker::RunMigration(
                                                   PendingCatalog *data) const {
  return CheckDatabaseSchemaCompatibility(data) &&
         BreakUpHardlinks(data);
}


bool
CommandMigrate::HardlinkRemovalMigrationWorker::CheckDatabaseSchemaCompatibility
                                                  (PendingCatalog *data) const {
  assert(data->old_catalog != NULL);
  assert(data->new_catalog == NULL);

  const catalog::CatalogDatabase &clg = data->old_catalog->database();
  return clg.schema_version() >= 2.4 - catalog::CatalogDatabase::kSchemaEpsilon;
}


bool CommandMigrate::HardlinkRemovalMigrationWorker::BreakUpHardlinks(
                                                   PendingCatalog *data) const {
  assert(data->old_catalog != NULL);
  assert(data->new_catalog == NULL);

  const catalog::CatalogDatabase &db =
                                     GetWritable(data->old_catalog)->database();

  if (!db.BeginTransaction()) {
    return false;
  }

  // CernVM-FS catalogs do not contain inodes directly but they are assigned by
  // the CVMFS catalog at runtime. Hardlinks are treated with so-called hardlink
  // group IDs to indicate hardlink relationships that need to be respected at
  // runtime by assigning identical inodes accordingly.
  //
  // This updates all directory entries of a given catalog that have a linkcount
  // greater than 1 and are flagged as a 'file'. Note: Symlinks are flagged both
  // as 'file' and as 'symlink', hence they are updated implicitly as well.
  //
  // The 'hardlinks' field in the catalog contains two 32 bit integers:
  //   * the linkcount in the lower 32 bits
  //   * the (so called) hardlink group ID in the higher 32 bits
  //
  // Files that have a linkcount of exactly 1 do not have any hardlinks and have
  // the (implicit) hardlink group ID '0'. Hence, 'hardlinks == 1' means that a
  // file doesn't have any hardlinks (linkcount = 1) and doesn't need treatment
  // here.
  //
  // Files that have hardlinks (linkcount > 1) will have a very large integer in
  // their 'hardlinks' field (hardlink group ID > 0 in higher 32 bits). Those
  // files will be treated by setting their 'hardlinks' field to 1, effectively
  // clearing all hardlink information from the directory entry.
  const std::string stmt = "UPDATE OR ABORT catalog "
                           "SET hardlinks = 1 "
                           "WHERE flags & :file_flag "
                           "  AND hardlinks > 1;";
  catalog::SqlCatalog hardlink_removal_sql(db, stmt);
  hardlink_removal_sql.BindInt64(1, catalog::SqlDirent::kFlagFile);
  hardlink_removal_sql.Execute();

  return db.CommitTransaction();
}

//------------------------------------------------------------------------------


bool CommandMigrate::BulkhashRemovalMigrationWorker::RunMigration(
                                                   PendingCatalog *data) const {
  return CheckDatabaseSchemaCompatibility(data) &&
         RemoveRedundantBulkHashes(data);
}


bool
CommandMigrate::BulkhashRemovalMigrationWorker::CheckDatabaseSchemaCompatibility
                                                  (PendingCatalog *data) const {
  assert(data->old_catalog != NULL);
  assert(data->new_catalog == NULL);

  const catalog::CatalogDatabase &clg = data->old_catalog->database();
  return clg.schema_version() >= 2.4 - catalog::CatalogDatabase::kSchemaEpsilon;
}


bool CommandMigrate::BulkhashRemovalMigrationWorker::RemoveRedundantBulkHashes(
                                                   PendingCatalog *data) const {
  assert(data->old_catalog != NULL);
  assert(data->new_catalog == NULL);

  const catalog::CatalogDatabase &db =
                                     GetWritable(data->old_catalog)->database();

  if (!db.BeginTransaction()) {
    return false;
  }

  // Regular files with both bulk hashes and chunked hashes can drop the bulk
  // hash since modern clients >= 2.1.7 won't require them
  const std::string stmt = "UPDATE OR ABORT catalog "
                           "SET hash = NULL "
                           "WHERE flags & :file_chunked_flag;";
  catalog::SqlCatalog bulkhash_removal_sql(db, stmt);
  bulkhash_removal_sql.BindInt64(1, catalog::SqlDirent::kFlagFileChunk);
  bulkhash_removal_sql.Execute();

  return db.CommitTransaction();
}


//------------------------------------------------------------------------------


CommandMigrate::StatsMigrationWorker::StatsMigrationWorker(
  const worker_context *context)
  : AbstractMigrationWorker<StatsMigrationWorker>(context)
{ }


bool CommandMigrate::StatsMigrationWorker::RunMigration(PendingCatalog *data)
  const
{
  return CheckDatabaseSchemaCompatibility(data) &&
         StartDatabaseTransaction(data) &&
         RepairStatisticsCounters(data) &&
         CommitDatabaseTransaction(data);
}


bool CommandMigrate::StatsMigrationWorker::CheckDatabaseSchemaCompatibility(
  PendingCatalog *data) const
{
  assert(data->old_catalog != NULL);
  assert(data->new_catalog == NULL);

  const catalog::CatalogDatabase &clg = data->old_catalog->database();
  if (clg.schema_version() < 2.5 - catalog::CatalogDatabase::kSchemaEpsilon) {
    Error("Given catalog schema is < 2.5.", data);
    return false;
  }

  if (clg.schema_revision() < 5) {
    Error("Given catalog revision is < 5", data);
    return false;
  }

  return true;
}


bool CommandMigrate::StatsMigrationWorker::StartDatabaseTransaction(
  PendingCatalog *data) const
{
  assert(!data->HasNew());
  GetWritable(data->old_catalog)->Transaction();
  return true;
}


bool CommandMigrate::StatsMigrationWorker::RepairStatisticsCounters(
  PendingCatalog *data) const
{
  assert(!data->HasNew());
  bool retval = false;
  const catalog::CatalogDatabase &writable =
    GetWritable(data->old_catalog)->database();

  // Aggregated the statistics counters of all nested catalogs
  // Note: we might need to wait until nested catalogs are sucessfully processed
  catalog::DeltaCounters stats_counters;
  PendingCatalogList::const_iterator i = data->nested_catalogs.begin();
  PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    const PendingCatalog *nested_catalog = *i;
    const catalog::DeltaCounters &s = nested_catalog->nested_statistics.Get();
    s.PopulateToParent(&stats_counters);
  }

  // Count various directory entry types in the catalog to fill up the catalog
  // statistics counters introduced in the current catalog schema
  catalog::SqlCatalog count_regular(writable,
    std::string("SELECT count(*), sum(size) FROM catalog ") +
    "WHERE flags & " + StringifyInt(catalog::SqlDirent::kFlagFile) +
    " AND NOT flags & " + StringifyInt(catalog::SqlDirent::kFlagLink) +
    " AND NOT flags & " + StringifyInt(catalog::SqlDirent::kFlagFileSpecial) +
    ";");
  catalog::SqlCatalog count_external(writable,
    std::string("SELECT count(*), sum(size) FROM catalog ") +
    "WHERE flags & " + StringifyInt(catalog::SqlDirent::kFlagFileExternal) +
    ";");
  catalog::SqlCatalog count_symlink(writable,
    std::string("SELECT count(*) FROM catalog ") +
    "WHERE flags & " + StringifyInt(catalog::SqlDirent::kFlagLink) + ";");
  catalog::SqlCatalog count_special(writable,
    std::string("SELECT count(*) FROM catalog ") +
    "WHERE flags & " + StringifyInt(catalog::SqlDirent::kFlagFileSpecial) +
    ";");
  catalog::SqlCatalog count_xattr(writable,
    std::string("SELECT count(*) FROM catalog ") +
    "WHERE xattr IS NOT NULL;");
  catalog::SqlCatalog count_chunk(writable,
    std::string("SELECT count(*), sum(size) FROM catalog ") +
    "WHERE flags & " + StringifyInt(catalog::SqlDirent::kFlagFileChunk) + ";");
  catalog::SqlCatalog count_dir(writable,
    std::string("SELECT count(*) FROM catalog ") +
    "WHERE flags & " + StringifyInt(catalog::SqlDirent::kFlagDir) + ";");
  catalog::SqlCatalog count_chunk_blobs(writable,
    "SELECT count(*) FROM chunks;");

  retval = count_regular.FetchRow() &&
           count_external.FetchRow() &&
           count_symlink.FetchRow() &&
           count_special.FetchRow() &&
           count_xattr.FetchRow() &&
           count_chunk.FetchRow() &&
           count_dir.FetchRow() &&
           count_chunk_blobs.FetchRow();
  if (!retval) {
    Error("Failed to collect catalog statistics", data);
    return false;
  }

  stats_counters.self.regular_files       = count_regular.RetrieveInt64(0);
  stats_counters.self.symlinks            = count_symlink.RetrieveInt64(0);
  stats_counters.self.specials            = count_special.RetrieveInt64(0);
  stats_counters.self.directories         = count_dir.RetrieveInt64(0);
  stats_counters.self.nested_catalogs     = data->nested_catalogs.size();
  stats_counters.self.chunked_files       = count_chunk.RetrieveInt64(0);
  stats_counters.self.file_chunks         = count_chunk_blobs.RetrieveInt64(0);
  stats_counters.self.file_size           = count_regular.RetrieveInt64(1);
  stats_counters.self.chunked_file_size   = count_chunk.RetrieveInt64(1);
  stats_counters.self.xattrs              = count_xattr.RetrieveInt64(0);
  stats_counters.self.externals           = count_external.RetrieveInt64(0);
  stats_counters.self.external_file_size  = count_external.RetrieveInt64(1);

  // Write back the generated statistics counters into the catalog database
  catalog::Counters counters;
  counters.ApplyDelta(stats_counters);
  retval = counters.InsertIntoDatabase(writable);
  if (!retval) {
    Error("Failed to write new statistics counters to database", data);
    return false;
  }

  // Push the generated statistics counters up to the parent catalog
  data->nested_statistics.Set(stats_counters);

  return true;
}


bool CommandMigrate::StatsMigrationWorker::CommitDatabaseTransaction(
  PendingCatalog *data) const
{
  assert(!data->HasNew());
  GetWritable(data->old_catalog)->Commit();
  return true;
}

}  // namespace swissknife

<MSG> Merge pull request #1518 from reneme/feature-lazy_sql_stmt

Feature: Lazy Initialisation of SQL statements

This allows derived `Sql` wrapper classes to opt for lazy initialisation of their prepared statements. We hope to safe memory and CPU cycles this way. For example each catalog opened for writing is preparing 17 SQL statements even though it might only ever need a small subset of them.

Furthermore this changes some low-hanging fruits in history_sql.cc to utilise the new `Sql::DeferredInit()` for demoing the API.
<DFF> @@ -1302,36 +1302,6 @@ void CommandMigrate::FixNestedCatalogTransitionPoint(
 }
 
 
-/**
- * Queries a single catalog and looks for DirectoryEntrys that have direct
- * children in the same catalog but are marked as 'nested catalog mountpoints'.
- * This is an inconsistent situation, since a mountpoint is supposed to be empty
- * and it's children are stored in the corresponding referenced nested catalog.
- *
- * Note: the user code needs to check if there is a corresponding nested catalog
- *       reference for the found dangling mountpoints. If so, we also have a
- *       bogus state, but it is not reliably fixable automatically. The child-
- *       DirectoryEntrys would be masked by the mounting nested catalog but it
- *       is not clear if we can simply delete them or if this would destroy data.
- */
-class SqlLookupDanglingMountpoints : public catalog::SqlLookup {
- public:
-  explicit SqlLookupDanglingMountpoints(
-    const catalog::CatalogDatabase &database)
-  {
-    const string statement =
-      "SELECT DISTINCT " +
-      GetFieldsToSelect(database.schema_version(), database.schema_revision()) +
-      " FROM catalog "
-      "JOIN catalog AS c2 "
-      "ON catalog.md5path_1 = c2.parent_1 AND catalog.md5path_2 = c2.parent_2 "
-      "WHERE catalog.flags & "
-      + StringifyInt(catalog::SqlDirent::kFlagDirNestedMountpoint) + ";";
-    const bool init_successful = Init(database.sqlite_db(), statement);
-    assert(init_successful);
-  }
-};
-
 bool CommandMigrate::MigrationWorker_20x::RemoveDanglingNestedMountpoints(
   PendingCatalog *data) const
 {
@@ -1356,7 +1326,7 @@ bool CommandMigrate::MigrationWorker_20x::RemoveDanglingNestedMountpoints(
 
   // Retrieve nested catalog mountpoints that have child entries directly inside
   // the current catalog (which is a malformed state)
-  SqlLookupDanglingMountpoints sql_dangling_mountpoints(writable);
+  catalog::SqlLookupDanglingMountpoints sql_dangling_mountpoints(writable);
   catalog::SqlDirentUpdate save_updated_mountpoint(writable);
 
   // go through the list of dangling nested catalog mountpoints and fix them

{'addition_count': 1, 'commit_subject': 'Merge pull request #1518 from reneme/feature-lazy_sql_stmt', 'deletion_count': 31, 'file_extension': '.cc', 'lang': 'cc', 'license': 'bsd-3-clause', 'repo_name': 'cvmfs/cvmfs'}