swissknife_migrate.cc

Go to the documentation of this file.

#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/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, "%s", 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->InitSignatureManager(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,
                                    uint64_t revision,
                                    time_t timestamp,
                                    shash::Any *history_hash) {
  const string filename_old = history_upstream_->filename();
  const 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;
  const 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 = 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("Updating 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
  {
    const 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
  const 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;
    {
      const LockGuard<PendingCatalogMap> guard(&pending_catalogs_);
      const 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();
  const 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 {
  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();
  const 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
  // currently 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();
  const 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();
  const PendingCatalogList::const_iterator iend = data->nested_catalogs.end();
  for (; i != iend; ++i) {
    delete *i;
  }

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


const float CommandMigrate::MigrationWorker_20x::kSchema = 2.5;
const unsigned CommandMigrate::MigrationWorker_20x::kSchemaRevision = 7;


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
    const 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;");
  const 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, NULL, "  // set empty mtimens
      "         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)
  PendingCatalogList::const_iterator i = data->nested_catalogs.begin();
  const 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 root_entry = nested_catalog->root_entry.Get();
    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();
  }

  return true;
}


bool CommandMigrate::MigrationWorker_20x::FixNestedCatalogTransitionPoints(
    PendingCatalog *data) const {
  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();
  const 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
    const 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 occurred
      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 successfully
  // processed
  catalog::DeltaCounters stats_counters;
  PendingCatalogList::const_iterator i = data->nested_catalogs.begin();
  const 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();
  const 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;
  }

  const 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 successfully
  // processed
  catalog::DeltaCounters stats_counters;
  PendingCatalogList::const_iterator i = data->nested_catalogs.begin();
  const 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 successfully
  // processed
  catalog::DeltaCounters stats_counters;
  PendingCatalogList::const_iterator i = data->nested_catalogs.begin();
  const 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