catalog_sql.cc

Go to the documentation of this file.

#include "catalog_sql.h"

#include <cstdlib>
#include <cstring>

#include "catalog.h"
#include "globals.h"
#include "util/logging.h"
#include "util/posix.h"
#include "xattr.h"

using namespace std;  // NOLINT

namespace catalog {

// ChangeLog
// 2.5 (Jun 26 2013 - Git: e79baec22c6abd6ddcdf8f8d7d33921027a052ab)
//     * add (backward compatible) schema revision - see below
//     * add statistics counters for chunked files
//       Note: this was retrofitted and needed a catalog migration step
//
// 2.4 (Aug 15 2012 - Git: 17de8fc782b5b8dc4404dda925627b5ec2b552e1)
// 2.3 (Aug 15 2012 - Git: ab77688cdb2f851af3fe983bf3694dc2465e65be)
// 2.2 (never existed)
// 2.1 (Aug  7 2012 - Git: beba36c12d2b1123ffbb169f865a861e570adc68)
//     * add 'chunks' table for file chunks
//     * add 'statistics' table for accumulative counters
//     * rename 'inode' field to 'hardlinks'
//       * containing both hardlink group ID and linkcount
//     * .cvmfscatalog files become first-class entries in the catalogs
//
// 2.0 (Aug  6 2012 - Git: c8a81ede603e57fbe4324b6ab6bc8c41e3a2fa5f)
//     * beginning of CernVM-FS 2.1.x branch ('modern' era)
//
// 1.x (earlier - code base still in SVN)
//     * pre-historic times
// 0.9 (some time 2011, artificial version)
//     * 1.0 catalogs that lack the SHA-1 value for nested catalogs
const float CatalogDatabase::kLatestSchema = 2.5;
const float CatalogDatabase::kLatestSupportedSchema = 2.5;  // + 1.X (r/o)

// ChangeLog
//   0 --> 1: (Jan  6 2014 - Git: 3667fe7a669d0d65e07275b753a7c6f23fc267df)
//            * add size column to nested catalog table,
//            * add schema_revision property
//   1 --> 2: (Jan 22 2014 - Git: 85e6680e52cfe56dc1213a5ad74a5cc62fd50ead):
//            * add xattr column to catalog table
//            * add self_xattr and subtree_xattr statistics counters
//   2 --> 3: (Sep 28 2015 - Git: f4171234b13ea448589820c1524ee52eae141bb4):
//            * add kFlagFileExternal to entries in catalog table
//            * add self_external and subtree_external statistics counters
//            * store compression algorithm in flags
//   3 --> 4: (Nov 11 2016 - Git):
//            * add kFlagDirBindMountpoint
//            * add kFlagHidden
//            * add table bind_mountpoints
//   4 --> 5: (Dec 07 2017):
//            * add kFlagFileSpecial (rebranded unused kFlagFileStat)
//            * add self_special and subtree_special statistics counters
//   5 --> 6: (Jul 01 2021):
//            * Add kFlagDirectIo
//   6 --> 7: (Feb 23 2024):
//            * Store nanosecond timestamps (mtimens column) in catalog table.
//              The mtimens column has only the nanosecond part of the timestamp
//              and may be NULL
const unsigned CatalogDatabase::kLatestSchemaRevision = 7;

bool CatalogDatabase::CheckSchemaCompatibility() {
  return !((schema_version() >= 2.0 - kSchemaEpsilon)
           && (!IsEqualSchema(schema_version(), kLatestSupportedSchema))
           && (!IsEqualSchema(schema_version(), 2.4)
               || !IsEqualSchema(kLatestSupportedSchema, 2.5)));
}


bool CatalogDatabase::LiveSchemaUpgradeIfNecessary() {
  assert(read_write());

  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 0)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (0 --> 1)");

    SqlCatalog sql_upgrade(*this, "ALTER TABLE nested_catalogs "
                                  "ADD size INTEGER;");
    if (!sql_upgrade.Execute()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade nested_catalogs");
      return false;
    }

    set_schema_revision(1);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }

  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 1)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (1 --> 2)");

    SqlCatalog sql_upgrade1(*this, "ALTER TABLE catalog ADD xattr BLOB;");
    SqlCatalog sql_upgrade2(
        *this,
        "INSERT INTO statistics (counter, value) VALUES ('self_xattr', 0);");
    SqlCatalog sql_upgrade3(
        *this,
        "INSERT INTO statistics (counter, value) VALUES ('subtree_xattr', 0);");
    if (!sql_upgrade1.Execute() || !sql_upgrade2.Execute()
        || !sql_upgrade3.Execute()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade catalogs (1 --> 2)");
      return false;
    }

    set_schema_revision(2);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }

  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 2)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (2 --> 3)");

    SqlCatalog sql_upgrade4(*this,
                            "INSERT INTO statistics (counter, value) VALUES "
                            "('self_external', 0);");
    SqlCatalog sql_upgrade5(*this,
                            "INSERT INTO statistics (counter, value) VALUES "
                            "('self_external_file_size', 0);");
    SqlCatalog sql_upgrade6(*this,
                            "INSERT INTO statistics (counter, value) VALUES "
                            "('subtree_external', 0);");
    SqlCatalog sql_upgrade7(*this,
                            "INSERT INTO statistics (counter, value) VALUES "
                            "('subtree_external_file_size', 0);");
    if (!sql_upgrade4.Execute() || !sql_upgrade5.Execute()
        || !sql_upgrade6.Execute() || !sql_upgrade7.Execute()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade catalogs (2 --> 3)");
      return false;
    }

    set_schema_revision(3);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }

  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 3)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (3 --> 4)");

    SqlCatalog sql_upgrade8(
        *this,
        "CREATE TABLE bind_mountpoints (path TEXT, sha1 TEXT, size INTEGER, "
        "CONSTRAINT pk_bind_mountpoints PRIMARY KEY (path));");
    if (!sql_upgrade8.Execute()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade catalogs (3 --> 4)");
      return false;
    }

    set_schema_revision(4);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }


  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 4)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (4 --> 5)");

    SqlCatalog sql_upgrade9(*this,
                            "INSERT INTO statistics (counter, value) VALUES "
                            "('self_special', 0);");
    SqlCatalog sql_upgrade10(*this,
                             "INSERT INTO statistics (counter, value) VALUES "
                             "('subtree_special', 0);");
    if (!sql_upgrade9.Execute() || !sql_upgrade10.Execute()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade catalogs (4 --> 5)");
      return false;
    }

    set_schema_revision(5);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }


  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 5)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (5 --> 6)");

    set_schema_revision(6);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }

  if (IsEqualSchema(schema_version(), 2.5) && (schema_revision() == 6)) {
    LogCvmfs(kLogCatalog, kLogDebug, "upgrading schema revision (6 --> 7)");

    SqlCatalog sql_upgrade1(*this, "ALTER TABLE catalog ADD mtimens INTEGER;");
    if (!sql_upgrade1.Execute()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade catalogs (6 --> 7)");
      return false;
    }

    set_schema_revision(7);
    if (!StoreSchemaRevision()) {
      LogCvmfs(kLogCatalog, kLogDebug, "failed to upgrade schema revision");
      return false;
    }
  }

  return true;
}


bool CatalogDatabase::CreateEmptyDatabase() {
  assert(read_write());

  // generate the catalog table and index structure
  const bool retval =
      SqlCatalog(*this,
                 "CREATE TABLE catalog "
                 "(md5path_1 INTEGER, md5path_2 INTEGER, parent_1 INTEGER, "
                 "parent_2 INTEGER,"
                 " hardlinks INTEGER, hash BLOB, size INTEGER, mode INTEGER, "
                 "mtime INTEGER,"
                 " mtimens INTEGER, flags INTEGER, name TEXT, symlink TEXT, "
                 "uid INTEGER,"
                 " gid INTEGER, xattr BLOB, "
                 " CONSTRAINT pk_catalog PRIMARY KEY (md5path_1, md5path_2));")
          .Execute()
      && SqlCatalog(*this,
                    "CREATE INDEX idx_catalog_parent "
                    "ON catalog (parent_1, parent_2);")
             .Execute()
      && SqlCatalog(*this,
                    "CREATE TABLE chunks "
                    "(md5path_1 INTEGER, md5path_2 INTEGER, offset INTEGER, "
                    "size INTEGER, "
                    " hash BLOB, "
                    " CONSTRAINT pk_chunks PRIMARY KEY (md5path_1, md5path_2, "
                    "offset, size), "
                    " FOREIGN KEY (md5path_1, md5path_2) REFERENCES "
                    "   catalog(md5path_1, md5path_2));")
             .Execute()
      && SqlCatalog(*this,
                    "CREATE TABLE nested_catalogs (path TEXT, sha1 TEXT, size "
                    "INTEGER, "
                    "CONSTRAINT pk_nested_catalogs PRIMARY KEY (path));")
             .Execute()
      &&
      // Bind mountpoints and nested catalogs are almost the same. We put them
      // in separate tables to
      //   - not confuse previous client versions, which would crash on bind
      //     mountpoints
      //   - prevent catalogs referenced as bind mountpoints from being
      //   replicated,
      //     which would cause exhaustive recursive catalog tree walking
      //   - don't walk into bind mountpoints in catalog traversal (e.g. GC)
      SqlCatalog(
          *this,
          "CREATE TABLE bind_mountpoints (path TEXT, sha1 TEXT, size INTEGER, "
          "CONSTRAINT pk_bind_mountpoints PRIMARY KEY (path));")
          .Execute()
      && SqlCatalog(*this,
                    "CREATE TABLE statistics (counter TEXT, value INTEGER, "
                    "CONSTRAINT pk_statistics PRIMARY KEY (counter));")
             .Execute();

  if (!retval) {
    PrintSqlError("failed to create catalog database tables.");
  }

  return retval;
}


bool CatalogDatabase::InsertInitialValues(const std::string &root_path,
                                          const bool volatile_content,
                                          const std::string &voms_authz,
                                          const DirectoryEntry &root_entry) {
  assert(read_write());
  bool retval = false;

  // Path hashes
  const shash::Md5 root_path_hash = shash::Md5(shash::AsciiPtr(root_path));
  const shash::Md5 root_parent_hash =
      (root_path == "") ? shash::Md5()
                        : shash::Md5(shash::AsciiPtr(GetParentPath(root_path)));

  // Start initial filling transaction
  retval = BeginTransaction();
  if (!retval) {
    PrintSqlError("failed to enter initial filling transaction");
    return false;
  }

  // Insert initial values to properties
  if (!this->SetProperty("revision", 0)) {
    PrintSqlError(
        "failed to insert default initial values into the newly created "
        "catalog tables.");
    return false;
  }

  if (volatile_content) {
    if (!this->SetProperty("volatile", 1)) {
      PrintSqlError("failed to insert volatile flag into the newly created "
                    "catalog tables.");
      return false;
    }
  }

  if (!voms_authz.empty()) {
    if (!SetVOMSAuthz(voms_authz)) {
      PrintSqlError("failed to insert VOMS authz flag into the newly created "
                    "catalog tables.");
      return false;
    }
  }

  // Create initial statistics counters
  catalog::Counters counters;

  // Insert root entry (when given)
  if (!root_entry.IsNegative()) {
    SqlDirentInsert sql_insert(*this);
    retval = sql_insert.BindPathHash(root_path_hash)
             && sql_insert.BindParentPathHash(root_parent_hash)
             && sql_insert.BindDirent(root_entry) && sql_insert.Execute();
    if (!retval) {
      PrintSqlError("failed to insert root entry into newly created catalog.");
      return false;
    }

    // account for the created root entry
    counters.self.directories = 1;
  }

  // Save initial statistics counters
  if (!counters.InsertIntoDatabase(*this)) {
    PrintSqlError("failed to insert initial catalog statistics counters.");
    return false;
  }

  // Insert root path (when given)
  if (!root_path.empty()) {
    if (!this->SetProperty("root_prefix", root_path)) {
      PrintSqlError(
          "failed to store root prefix in the newly created catalog.");
      return false;
    }
  }

  // Set creation timestamp
  if (!this->SetProperty("last_modified", static_cast<uint64_t>(time(NULL)))) {
    PrintSqlError("failed to store creation timestamp in the new catalog.");
    return false;
  }

  // Commit initial filling transaction
  retval = CommitTransaction();
  if (!retval) {
    PrintSqlError("failed to commit initial filling transaction");
    return false;
  }

  return true;
}


bool CatalogDatabase::SetVOMSAuthz(const std::string &voms_authz) {
  return this->SetProperty("voms_authz", voms_authz);
}


double CatalogDatabase::GetRowIdWasteRatio() const {
  SqlCatalog rowid_waste_ratio_query(
      *this,
      "SELECT 1.0 - CAST(COUNT(*) AS DOUBLE) / MAX(rowid) "
      "AS ratio FROM catalog;");
  const bool retval = rowid_waste_ratio_query.FetchRow();
  assert(retval);

  return rowid_waste_ratio_query.RetrieveDouble(0);
}

bool CatalogDatabase::CompactDatabase() const {
  assert(read_write());

  return SqlCatalog(*this, "PRAGMA foreign_keys = OFF;").Execute()
         && BeginTransaction()
         && SqlCatalog(*this, "CREATE TEMPORARY TABLE duplicate AS "
                              "  SELECT * FROM catalog "
                              "  ORDER BY rowid ASC;")
                .Execute()
         && SqlCatalog(*this, "DELETE FROM catalog;").Execute()
         && SqlCatalog(*this, "INSERT INTO catalog "
                              "  SELECT * FROM duplicate "
                              "  ORDER BY rowid")
                .Execute()
         && SqlCatalog(*this, "DROP TABLE duplicate;").Execute()
         && CommitTransaction()
         && SqlCatalog(*this, "PRAGMA foreign_keys = ON;").Execute();
}


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


unsigned SqlDirent::CreateDatabaseFlags(const DirectoryEntry &entry) const {
  unsigned int database_flags = 0;

  if (entry.IsNestedCatalogRoot())
    database_flags |= kFlagDirNestedRoot;
  else if (entry.IsNestedCatalogMountpoint())
    database_flags |= kFlagDirNestedMountpoint;
  else if (entry.IsBindMountpoint())
    database_flags |= kFlagDirBindMountpoint;

  if (entry.IsDirectory()) {
    database_flags |= kFlagDir;
  } else if (entry.IsLink()) {
    database_flags |= kFlagFile | kFlagLink;
  } else if (entry.IsSpecial()) {
    database_flags |= kFlagFile | kFlagFileSpecial;
  } else {
    database_flags |= kFlagFile;
    database_flags |= entry.compression_algorithm() << kFlagPosCompression;
    if (entry.IsChunkedFile())
      database_flags |= kFlagFileChunk;
    if (entry.IsExternalFile())
      database_flags |= kFlagFileExternal;
    if (entry.IsDirectIo())
      database_flags |= kFlagDirectIo;
  }

  if (!entry.checksum_ptr()->IsNull() || entry.IsChunkedFile())
    StoreHashAlgorithm(entry.checksum_ptr()->algorithm, &database_flags);

  if (entry.IsHidden())
    database_flags |= kFlagHidden;

  return database_flags;
}


void SqlDirent::StoreHashAlgorithm(const shash::Algorithms algo,
                                   unsigned *flags) const {
  assert(algo != shash::kAny);
  // Md5 unusable for content hashes
  *flags |= (algo - 1) << kFlagPosHash;
}


shash::Algorithms SqlDirent::RetrieveHashAlgorithm(const unsigned flags) const {
  unsigned in_flags = ((7 << kFlagPosHash) & flags) >> kFlagPosHash;
  // Skip Md5
  in_flags++;
  assert(in_flags < shash::kAny);
  return static_cast<shash::Algorithms>(in_flags);
}


zlib::Algorithms SqlDirent::RetrieveCompressionAlgorithm(
    const unsigned flags) const {
  // 3 bits, so use 7 (111) to only pull out the flags we want
  const unsigned in_flags =
      ((7 << kFlagPosCompression) & flags) >> kFlagPosCompression;
  return static_cast<zlib::Algorithms>(in_flags);
}


uint32_t SqlDirent::Hardlinks2Linkcount(const uint64_t hardlinks) const {
  return (hardlinks << 32) >> 32;
}


uint32_t SqlDirent::Hardlinks2HardlinkGroup(const uint64_t hardlinks) const {
  return hardlinks >> 32;
}


uint64_t SqlDirent::MakeHardlinks(const uint32_t hardlink_group,
                                  const uint32_t linkcount) const {
  assert(linkcount > 0);
  return (static_cast<uint64_t>(hardlink_group) << 32) | linkcount;
}


void SqlDirent::ExpandSymlink(LinkString *raw_symlink) const {
  const char *c = raw_symlink->GetChars();
  const char *cEnd = c + raw_symlink->GetLength();
  for (; c < cEnd; ++c) {
    if (*c == '$')
      goto expand_symlink;
  }
  return;

expand_symlink:
  LinkString result;
  for (c = raw_symlink->GetChars(); c < cEnd; ++c) {
    if ((*c == '$') && (c < cEnd - 2) && (*(c + 1) == '(')) {
      c += 2;
      const char *rpar = c;
      while (rpar < cEnd) {
        if (*rpar == ')')
          goto expand_symlink_getenv;
        rpar++;
      }
      // right parenthesis missing
      result.Append("$(", 2);
      result.Append(c, 1);
      continue;

    expand_symlink_getenv:
      // Check for default value
      const char *default_separator = c;
      const char *default_value = rpar;
      while (default_separator != rpar) {
        if ((*default_separator == ':') && (*(default_separator + 1) == '-')) {
          default_value = default_separator + 2;
          break;
        }
        default_separator++;
      }

      const unsigned environ_var_length = default_separator - c;
      char environ_var[environ_var_length + 1];
      environ_var[environ_var_length] = '\0';
      memcpy(environ_var, c, environ_var_length);
      const char *environ_value = getenv(environ_var);  // Don't free!
      if (environ_value) {
        result.Append(environ_value, strlen(environ_value));
      } else {
        const unsigned default_length = rpar - default_value;
        result.Append(default_value, default_length);
      }
      c = rpar;
      continue;
    }
    result.Append(c, 1);
  }
  raw_symlink->Assign(result);
  return;
}


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


bool SqlDirentWrite::BindDirentFields(const int hash_idx,
                                      const int hardlinks_idx,
                                      const int size_idx,
                                      const int mode_idx,
                                      const int mtime_idx,
                                      const int mtimens_idx,
                                      const int flags_idx,
                                      const int name_idx,
                                      const int symlink_idx,
                                      const int uid_idx,
                                      const int gid_idx,
                                      const DirectoryEntry &entry) {
  const uint64_t hardlinks = MakeHardlinks(entry.hardlink_group_,
                                           entry.linkcount_);

  bool result = BindHashBlob(hash_idx, entry.checksum_)
                && BindInt64(hardlinks_idx, hardlinks)
                && BindInt64(size_idx, entry.size_)
                && BindInt(mode_idx, entry.mode_)
                && BindInt64(uid_idx, entry.uid_)
                && BindInt64(gid_idx, entry.gid_)
                && BindInt64(mtime_idx, entry.mtime_)
                && BindInt(flags_idx, CreateDatabaseFlags(entry))
                && BindText(name_idx, entry.name_.GetChars(),
                            static_cast<int>(entry.name_.GetLength()))
                && BindText(symlink_idx, entry.symlink_.GetChars(),
                            static_cast<int>(entry.symlink_.GetLength()));
  if (entry.HasMtimeNs()) {
    result &= BindInt(mtimens_idx, entry.mtime_ns_);
  } else {
    result &= BindNull(mtimens_idx);
  }

  return result;
}


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


SqlListContentHashes::SqlListContentHashes(const CatalogDatabase &database) {
  static const char *stmt_lt_2_4 = "SELECT hash, flags, 0 "
                                   "  FROM catalog "
                                   "  WHERE length(hash) > 0;";

  static const char
      *stmt_ge_2_4 = "SELECT hash, flags, 0 "
                     "  FROM catalog "
                     "  WHERE (length(catalog.hash) > 0) AND "
                     "        ((flags & 128) = 0) "  // kFlagFileExternal
                     "UNION "
                     "SELECT chunks.hash, catalog.flags, 1 "
                     "  FROM catalog "
                     "  LEFT JOIN chunks "
                     "  ON catalog.md5path_1 = chunks.md5path_1 AND "
                     "     catalog.md5path_2 = chunks.md5path_2 "
                     "  WHERE (catalog.flags & 128) = 0;";  // kFlagFileExternal

  if (database.schema_version() < 2.4 - CatalogDatabase::kSchemaEpsilon) {
    DeferredInit(database.sqlite_db(), stmt_lt_2_4);
  } else {
    DeferredInit(database.sqlite_db(), stmt_ge_2_4);
  }
}


shash::Any SqlListContentHashes::GetHash() const {
  const unsigned int db_flags = RetrieveInt(1);
  const shash::Algorithms hash_algorithm = RetrieveHashAlgorithm(db_flags);
  shash::Any hash = RetrieveHashBlob(0, hash_algorithm);
  if (RetrieveInt(2) == 1) {
    hash.suffix = shash::kSuffixPartial;
  }

  return hash;
}


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

#define DB_FIELDS_LT_V2_1                                       \
  "catalog.hash,       catalog.inode,      catalog.size, "      \
  "catalog.mode,       catalog.mtime,      catalog.flags, "     \
  "catalog.name,       catalog.symlink,    catalog.md5path_1, " \
  "catalog.md5path_2,  catalog.parent_1,   catalog.parent_2, "  \
  "catalog.rowid"
#define DB_FIELDS_GE_V2_1_LT_R2                                 \
  "catalog.hash,       catalog.hardlinks,  catalog.size, "      \
  "catalog.mode,       catalog.mtime,      catalog.flags, "     \
  "catalog.name,       catalog.symlink,    catalog.md5path_1, " \
  "catalog.md5path_2,  catalog.parent_1,   catalog.parent_2, "  \
  "catalog.rowid,      catalog.uid,        catalog.gid, "       \
  "0, NULL"
#define DB_FIELDS_GE_V2_1_LT_R7                                 \
  "catalog.hash,       catalog.hardlinks,  catalog.size, "      \
  "catalog.mode,       catalog.mtime,      catalog.flags, "     \
  "catalog.name,       catalog.symlink,    catalog.md5path_1, " \
  "catalog.md5path_2,  catalog.parent_1,   catalog.parent_2, "  \
  "catalog.rowid,      catalog.uid,        catalog.gid, "       \
  "catalog.xattr IS NOT NULL, NULL"
#define DB_FIELDS_GE_V2_1_GE_R7                                 \
  "catalog.hash,       catalog.hardlinks,  catalog.size, "      \
  "catalog.mode,       catalog.mtime,      catalog.flags, "     \
  "catalog.name,       catalog.symlink,    catalog.md5path_1, " \
  "catalog.md5path_2,  catalog.parent_1,   catalog.parent_2, "  \
  "catalog.rowid,      catalog.uid,        catalog.gid, "       \
  "catalog.xattr IS NOT NULL, catalog.mtimens"

#define MAKE_STATEMENT(STMT_TMPL, REV)                                \
  static const std::string REV = ReplaceAll(STMT_TMPL, "@DB_FIELDS@", \
                                            DB_FIELDS_##REV)

#define MAKE_STATEMENTS(STMT_TMPL)          \
  MAKE_STATEMENT(STMT_TMPL, LT_V2_1);       \
  MAKE_STATEMENT(STMT_TMPL, GE_V2_1_LT_R2); \
  MAKE_STATEMENT(STMT_TMPL, GE_V2_1_LT_R7); \
  MAKE_STATEMENT(STMT_TMPL, GE_V2_1_GE_R7)

#define DEFERRED_INIT(DB, REV) DeferredInit((DB).sqlite_db(), (REV).c_str())

#define DEFERRED_INITS(DB)                                             \
  if ((DB).schema_version() < 2.1 - CatalogDatabase::kSchemaEpsilon) { \
    DEFERRED_INIT((DB), LT_V2_1);                                      \
  } else if ((DB).schema_revision() < 2) {                             \
    DEFERRED_INIT((DB), GE_V2_1_LT_R2);                                \
  } else if ((DB).schema_revision() < 7) {                             \
    DEFERRED_INIT((DB), GE_V2_1_LT_R7);                                \
  } else {                                                             \
    DEFERRED_INIT((DB), GE_V2_1_GE_R7);                                \
  }


shash::Md5 SqlLookup::GetPathHash() const { return RetrieveMd5(8, 9); }


shash::Md5 SqlLookup::GetParentPathHash() const { return RetrieveMd5(10, 11); }


DirectoryEntry SqlLookup::GetDirent(const Catalog *catalog,
                                    const bool expand_symlink) const {
  DirectoryEntry result;

  const unsigned database_flags = RetrieveInt(5);
  result.is_nested_catalog_root_ = (database_flags & kFlagDirNestedRoot);
  result.is_nested_catalog_mountpoint_ = (database_flags
                                          & kFlagDirNestedMountpoint);
  const char *name = reinterpret_cast<const char *>(RetrieveText(6));
  const char *symlink = reinterpret_cast<const char *>(RetrieveText(7));

  // Retrieve the hardlink information from the hardlinks database field
  if (catalog->schema() < 2.1 - CatalogDatabase::kSchemaEpsilon) {
    result.linkcount_ = 1;
    result.hardlink_group_ = 0;
    result.inode_ = catalog->GetMangledInode(RetrieveInt64(12), 0);
    result.is_chunked_file_ = false;
    result.has_xattrs_ = false;
    result.checksum_ = RetrieveHashBlob(0, shash::kSha1);
    result.uid_ = g_uid;
    result.gid_ = g_gid;
  } else {
    const uint64_t hardlinks = RetrieveInt64(1);
    result.linkcount_ = Hardlinks2Linkcount(hardlinks);
    result.hardlink_group_ = Hardlinks2HardlinkGroup(hardlinks);
    result.inode_ = catalog->GetMangledInode(RetrieveInt64(12),
                                             result.hardlink_group_);
    result.is_bind_mountpoint_ = (database_flags & kFlagDirBindMountpoint);
    result.is_chunked_file_ = (database_flags & kFlagFileChunk);
    result.is_hidden_ = (database_flags & kFlagHidden);
    result.is_direct_io_ = (database_flags & kFlagDirectIo);
    result.is_external_file_ = (database_flags & kFlagFileExternal);
    result.has_xattrs_ = RetrieveInt(15) != 0;
    result.mtime_ns_ = RetrieveNullableInt(16, -1);
    result.checksum_ = RetrieveHashBlob(0,
                                        RetrieveHashAlgorithm(database_flags));
    result.compression_algorithm_ = RetrieveCompressionAlgorithm(
        database_flags);

    if (g_claim_ownership) {
      result.uid_ = g_uid;
      result.gid_ = g_gid;
    } else {
      result.uid_ = catalog->MapUid(RetrieveInt64(13));
      result.gid_ = catalog->MapGid(RetrieveInt64(14));
    }
  }

  result.mode_ = RetrieveInt(3);
  result.size_ = RetrieveInt64(2);
  result.mtime_ = RetrieveInt64(4);
  result.name_.Assign(name, strlen(name));
  result.symlink_.Assign(symlink, strlen(symlink));
  if (expand_symlink && !g_raw_symlinks)
    ExpandSymlink(&result.symlink_);

  if (g_world_readable) {
    if (S_ISDIR(result.mode_)) {
      result.mode_ |= 0555;
    } else {
      result.mode_ |= 0444;
    }
  }

  return result;
}


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


SqlListing::SqlListing(const CatalogDatabase &database) {
  MAKE_STATEMENTS("SELECT @DB_FIELDS@ FROM catalog "
                  "WHERE (parent_1 = :p_1) AND (parent_2 = :p_2);");
  DEFERRED_INITS(database);
}


bool SqlListing::BindPathHash(const struct shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


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


SqlLookupPathHash::SqlLookupPathHash(const CatalogDatabase &database) {
  MAKE_STATEMENTS("SELECT @DB_FIELDS@ FROM catalog "
                  "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2);");
  DEFERRED_INITS(database);
}

bool SqlLookupPathHash::BindPathHash(const struct shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


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


SqlLookupInode::SqlLookupInode(const CatalogDatabase &database) {
  MAKE_STATEMENTS("SELECT @DB_FIELDS@ FROM catalog WHERE rowid = :rowid;");
  DEFERRED_INITS(database);
}


bool SqlLookupInode::BindRowId(const uint64_t inode) {
  return BindInt64(1, inode);
}


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


SqlLookupDanglingMountpoints::SqlLookupDanglingMountpoints(
    const catalog::CatalogDatabase &database) {
  MAKE_STATEMENTS("SELECT DISTINCT @DB_FIELDS@ FROM catalog "
                  "JOIN catalog AS c2 "
                  "ON catalog.md5path_1 = c2.parent_1 AND "
                  "   catalog.md5path_2 = c2.parent_2 "
                  "WHERE catalog.flags & :nested_mountpoint_flag");
  DEFERRED_INITS(database);

  // this pretty much removes the advantage of a deferred init but the statement
  // is anyway only used directly.
  const bool success = BindInt64(1, SqlDirent::kFlagDirNestedMountpoint);
  assert(success);
}


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


SqlDirentTouch::SqlDirentTouch(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "UPDATE catalog "
               "SET hash = :hash, size = :size, mode = :mode, mtime = :mtime, "
               //            1             2             3               4
               "name = :name, symlink = :symlink, uid = :uid, gid = :gid, "
               "xattr = :xattr, "
               //        5                6               7           8 9
               "mtimens = :mtimens "
               //             10
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2);");
  //                    11                       12
}


bool SqlDirentTouch::BindDirentBase(const DirectoryEntryBase &entry) {
  bool result = BindHashBlob(1, entry.checksum_) && BindInt64(2, entry.size_)
                && BindInt(3, entry.mode_) && BindInt64(4, entry.mtime_)
                && BindText(5, entry.name_.GetChars(), entry.name_.GetLength())
                && BindText(6, entry.symlink_.GetChars(),
                            entry.symlink_.GetLength())
                && BindInt64(7, entry.uid_) && BindInt64(8, entry.gid_);
  if (entry.HasMtimeNs()) {
    result &= BindInt(10, entry.mtime_ns_);
  } else {
    result &= BindNull(10);
  }
  return result;
}


bool SqlDirentTouch::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(11, 12, hash);
}


bool SqlDirentTouch::BindXattr(const XattrList &xattrs) {
  unsigned char *packed_xattrs;
  unsigned size;
  xattrs.Serialize(&packed_xattrs, &size);
  if (packed_xattrs == NULL)
    return BindNull(9);
  return BindBlobTransient(9, packed_xattrs, size);
}


bool SqlDirentTouch::BindXattrEmpty() { return BindNull(9); }


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


SqlNestedCatalogLookup::SqlNestedCatalogLookup(
    const CatalogDatabase &database) {
  // We cannot access nested catalogs where the content hash is missing
  static const char *stmt_0_9 = "SELECT '', 0 FROM nested_catalogs;";
  static const char *stmt_2_5_ge_4 =
      "SELECT sha1, size FROM nested_catalogs WHERE path=:path "
      "UNION ALL SELECT sha1, size FROM bind_mountpoints WHERE path=:path;";
  static const char *stmt_2_5_ge_1_lt_4 = "SELECT sha1, size FROM "
                                          "nested_catalogs WHERE path=:path;";
  // Internally converts NULL to 0 for size
  static const char
      *stmt_2_5_lt_1 = "SELECT sha1, 0 FROM nested_catalogs WHERE path=:path;";

  if (database.IsEqualSchema(database.schema_version(), 2.5)
      && (database.schema_revision() >= 4)) {
    DeferredInit(database.sqlite_db(), stmt_2_5_ge_4);
  } else if (database.IsEqualSchema(database.schema_version(), 2.5)
             && (database.schema_revision() >= 1)) {
    DeferredInit(database.sqlite_db(), stmt_2_5_ge_1_lt_4);
  } else {
    if (database.IsEqualSchema(database.schema_version(), 0.9)) {
      DeferredInit(database.sqlite_db(), stmt_0_9);
    } else {
      DeferredInit(database.sqlite_db(), stmt_2_5_lt_1);
    }
  }
}


bool SqlNestedCatalogLookup::BindSearchPath(const PathString &path) {
  return BindText(1, path.GetChars(), path.GetLength());
}


shash::Any SqlNestedCatalogLookup::GetContentHash() const {
  const string hash = string(reinterpret_cast<const char *>(RetrieveText(0)));
  return (hash.empty())
             ? shash::Any(shash::kAny)
             : shash::MkFromHexPtr(shash::HexPtr(hash), shash::kSuffixCatalog);
}


uint64_t SqlNestedCatalogLookup::GetSize() const { return RetrieveInt64(1); }


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


SqlNestedCatalogListing::SqlNestedCatalogListing(
    const CatalogDatabase &database) {
  // We cannot access nested catalogs where the content hash is missing
  static const char *stmt_0_9 = "SELECT '', '', 0 FROM nested_catalogs;";
  static const char *stmt_2_5_ge_4 =
      "SELECT path, sha1, size FROM nested_catalogs "
      "UNION ALL SELECT path, sha1, size FROM bind_mountpoints;";
  static const char
      *stmt_2_5_ge_1_lt_4 = "SELECT path, sha1, size FROM nested_catalogs;";
  // Internally converts NULL to 0 for size
  static const char
      *stmt_2_5_lt_1 = "SELECT path, sha1, 0 FROM nested_catalogs;";

  if (database.IsEqualSchema(database.schema_version(), 2.5)
      && (database.schema_revision() >= 4)) {
    DeferredInit(database.sqlite_db(), stmt_2_5_ge_4);
  } else if (database.IsEqualSchema(database.schema_version(), 2.5)
             && (database.schema_revision() >= 1)) {
    DeferredInit(database.sqlite_db(), stmt_2_5_ge_1_lt_4);
  } else {
    if (database.IsEqualSchema(database.schema_version(), 0.9)) {
      DeferredInit(database.sqlite_db(), stmt_0_9);
    } else {
      DeferredInit(database.sqlite_db(), stmt_2_5_lt_1);
    }
  }
}


PathString SqlNestedCatalogListing::GetPath() const {
  const char *path = reinterpret_cast<const char *>(RetrieveText(0));
  return PathString(path, strlen(path));
}


shash::Any SqlNestedCatalogListing::GetContentHash() const {
  const string hash = string(reinterpret_cast<const char *>(RetrieveText(1)));
  return (hash.empty())
             ? shash::Any(shash::kAny)
             : shash::MkFromHexPtr(shash::HexPtr(hash), shash::kSuffixCatalog);
}


uint64_t SqlNestedCatalogListing::GetSize() const { return RetrieveInt64(2); }


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


SqlOwnNestedCatalogListing::SqlOwnNestedCatalogListing(
    const CatalogDatabase &database) {
  // We cannot access nested catalogs where the content hash is missing
  static const char *stmt_0_9 = "SELECT '', '', 0 FROM nested_catalogs;";
  static const char
      *stmt_2_5_ge_1 = "SELECT path, sha1, size FROM nested_catalogs;";
  // Internally converts NULL to 0 for size
  static const char
      *stmt_2_5_lt_1 = "SELECT path, sha1, 0 FROM nested_catalogs;";

  if (database.IsEqualSchema(database.schema_version(), 2.5)
      && (database.schema_revision() >= 1)) {
    DeferredInit(database.sqlite_db(), stmt_2_5_ge_1);
  } else {
    if (database.IsEqualSchema(database.schema_version(), 0.9)) {
      DeferredInit(database.sqlite_db(), stmt_0_9);
    } else {
      DeferredInit(database.sqlite_db(), stmt_2_5_lt_1);
    }
  }
}


PathString SqlOwnNestedCatalogListing::GetPath() const {
  const char *path = reinterpret_cast<const char *>(RetrieveText(0));
  return PathString(path, strlen(path));
}


shash::Any SqlOwnNestedCatalogListing::GetContentHash() const {
  const string hash = string(reinterpret_cast<const char *>(RetrieveText(1)));
  return (hash.empty())
             ? shash::Any(shash::kAny)
             : shash::MkFromHexPtr(shash::HexPtr(hash), shash::kSuffixCatalog);
}


uint64_t SqlOwnNestedCatalogListing::GetSize() const {
  return RetrieveInt64(2);
}


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


SqlDirentInsert::SqlDirentInsert(const CatalogDatabase &database) {
  DeferredInit(
      database.sqlite_db(),
      "INSERT INTO catalog "
      "(md5path_1, md5path_2, parent_1, parent_2, hash, hardlinks, size, mode,"
      //    1           2         3         4       5       6        7     8
      "mtime, flags, name, symlink, uid, gid, xattr, mtimens) "
      // 9,     10    11     12     13   14   15     16
      "VALUES (:md5_1, :md5_2, :p_1, :p_2, :hash, :links, :size, :mode, :mtime,"
      " :flags, :name, :symlink, :uid, :gid, :xattr, :mtimens);");
}


bool SqlDirentInsert::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


bool SqlDirentInsert::BindParentPathHash(const shash::Md5 &hash) {
  return BindMd5(3, 4, hash);
}


bool SqlDirentInsert::BindDirent(const DirectoryEntry &entry) {
  return BindDirentFields(5, 6, 7, 8, 9, 16, 10, 11, 12, 13, 14, entry);
}


bool SqlDirentInsert::BindXattr(const XattrList &xattrs) {
  unsigned char *packed_xattrs;
  unsigned size;
  xattrs.Serialize(&packed_xattrs, &size);
  if (packed_xattrs == NULL)
    return BindNull(15);
  return BindBlobTransient(15, packed_xattrs, size);
}


bool SqlDirentInsert::BindXattrEmpty() { return BindNull(15); }


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


SqlDirentUpdate::SqlDirentUpdate(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "UPDATE catalog "
               "SET hash = :hash, size = :size, mode = :mode, mtime = :mtime, "
               //            1             2             3               4
               "flags = :flags, name = :name, symlink = :symlink, hardlinks = "
               ":hardlinks, "
               //          5             6                  7                8
               "uid = :uid, gid = :gid, mtimens = :mtimens "
               //          9           10             11
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2);");
  //                     12                       13
}


bool SqlDirentUpdate::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(12, 13, hash);
}


bool SqlDirentUpdate::BindDirent(const DirectoryEntry &entry) {
  return BindDirentFields(1, 8, 2, 3, 4, 11, 5, 6, 7, 9, 10, entry);
}


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


SqlDirentUnlink::SqlDirentUnlink(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "DELETE FROM catalog "
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2);");
}

bool SqlDirentUnlink::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


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


SqlIncLinkcount::SqlIncLinkcount(const CatalogDatabase &database) {
  // This command changes the linkcount of a whole hardlink group at once!
  // We can do this, since the 'hardlinks'-field contains the hardlink group ID
  // in the higher 32bit as well as the 'linkcount' in the lower 32bit.
  // This field will be equal for all entries belonging to the same hardlink
  // group while adding/subtracting small values from it will only effect the
  // linkcount in the lower 32bit.
  // Take a deep breath!
  DeferredInit(database.sqlite_db(),
               "UPDATE catalog SET hardlinks = hardlinks + :delta "
               "WHERE hardlinks = (SELECT hardlinks from catalog "
               "WHERE md5path_1 = :md5_1 AND md5path_2 = :md5_2);");
}


bool SqlIncLinkcount::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(2, 3, hash);
}


bool SqlIncLinkcount::BindDelta(const int delta) { return BindInt(1, delta); }


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


SqlChunkInsert::SqlChunkInsert(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "INSERT INTO chunks (md5path_1, md5path_2, offset, size, hash) "
               //                       1          2        3      4     5
               "VALUES (:md5_1, :md5_2, :offset, :size, :hash);");
}


bool SqlChunkInsert::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


bool SqlChunkInsert::BindFileChunk(const FileChunk &chunk) {
  return BindInt64(3, chunk.offset()) && BindInt64(4, chunk.size())
         && BindHashBlob(5, chunk.content_hash());
}


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


SqlChunksRemove::SqlChunksRemove(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "DELETE FROM chunks "
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2);");
}


bool SqlChunksRemove::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


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


SqlChunksListing::SqlChunksListing(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "SELECT offset, size, hash FROM chunks "
               //         0      1     2
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2) "
               //                    1                          2
               "ORDER BY offset ASC;");
}


bool SqlChunksListing::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


FileChunk SqlChunksListing::GetFileChunk(
    const shash::Algorithms interpret_hash_as) const {
  return FileChunk(
      RetrieveHashBlob(2, interpret_hash_as, shash::kSuffixPartial),
      RetrieveInt64(0),
      RetrieveInt64(1));
}


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


SqlChunksCount::SqlChunksCount(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "SELECT count(*) FROM chunks "
               //         0
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2)");
  //                    1                          2
}


bool SqlChunksCount::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


int SqlChunksCount::GetChunkCount() const { return RetrieveInt64(0); }


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


SqlMaxHardlinkGroup::SqlMaxHardlinkGroup(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(), "SELECT max(hardlinks) FROM catalog;");
}


uint32_t SqlMaxHardlinkGroup::GetMaxGroupId() const {
  return RetrieveInt64(0) >> 32;
}


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


SqlGetCounter::SqlGetCounter(const CatalogDatabase &database) {
  static const char
      *stmt_ge_2_4 = "SELECT value from statistics WHERE counter = :counter;";
  static const char *stmt_lt_2_4 = "SELECT 0;";

  if (database.schema_version() >= 2.4 - CatalogDatabase::kSchemaEpsilon) {
    compat_ = false;
    DeferredInit(database.sqlite_db(), stmt_ge_2_4);
  } else {
    compat_ = true;
    DeferredInit(database.sqlite_db(), stmt_lt_2_4);
  }
}


bool SqlGetCounter::BindCounter(const std::string &counter) {
  if (compat_)
    return true;
  return BindText(1, counter);
}


uint64_t SqlGetCounter::GetCounter() const {
  if (compat_)
    return 0;
  return RetrieveInt64(0);
}


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


SqlUpdateCounter::SqlUpdateCounter(const CatalogDatabase &database) {
  DeferredInit(
      database.sqlite_db(),
      "UPDATE statistics SET value=value+:val WHERE counter=:counter;");
}


bool SqlUpdateCounter::BindCounter(const std::string &counter) {
  return BindText(2, counter);
}


bool SqlUpdateCounter::BindDelta(const int64_t delta) {
  return BindInt64(1, delta);
}


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


SqlCreateCounter::SqlCreateCounter(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "INSERT OR REPLACE INTO statistics (counter, value) "
               "VALUES (:counter, :value);");
}


bool SqlCreateCounter::BindCounter(const std::string &counter) {
  return BindText(1, counter);
}


bool SqlCreateCounter::BindInitialValue(const int64_t value) {
  return BindInt64(2, value);
}


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


SqlAllChunks::SqlAllChunks(const CatalogDatabase &database) {
  const int hash_mask = 7 << SqlDirent::kFlagPosHash;
  const string flags2hash = " ((flags&" + StringifyInt(hash_mask) + ") >> " +
                            StringifyInt(SqlDirent::kFlagPosHash) +
                            ")+1 AS hash_algorithm ";

  const int compression_mask = 7 << SqlDirent::kFlagPosCompression;
  const string flags2compression =
      " ((flags&" + StringifyInt(compression_mask) + ") >> " +
      StringifyInt(SqlDirent::kFlagPosCompression) + ") " +
      "AS compression_algorithm ";

  // TODO(reneme): this depends on shash::kSuffix* being a char!
  //               it should be more generic or replaced entirely
  // TODO(reneme): this is practically the same as SqlListContentHashes and
  //               should be consolidated
  string sql = "SELECT DISTINCT hash, "
               "CASE WHEN flags & "
               + StringifyInt(SqlDirent::kFlagFile) + " THEN "
               + StringifyInt(shash::kSuffixNone) + " " + "WHEN flags & "
               + StringifyInt(SqlDirent::kFlagDir) + " THEN "
               + StringifyInt(shash::kSuffixMicroCatalog) + " END "
               + "AS chunk_type, " + flags2hash + "," + flags2compression
               + "FROM catalog WHERE (hash IS NOT NULL) AND "
                 "(flags & "
               + StringifyInt(SqlDirent::kFlagFileExternal) + " = 0)";
  if (database.schema_version() >= 2.4 - CatalogDatabase::kSchemaEpsilon) {
    sql += " UNION "
           "SELECT DISTINCT chunks.hash, "
           + StringifyInt(shash::kSuffixPartial) + ", " + flags2hash + ","
           + flags2compression
           + "FROM chunks, catalog WHERE "
             "chunks.md5path_1=catalog.md5path_1 AND "
             "chunks.md5path_2=catalog.md5path_2 AND "
             "(catalog.flags & "
           + StringifyInt(SqlDirent::kFlagFileExternal) + " = 0)";
  }
  sql += ";";
  Init(database.sqlite_db(), sql);
}


bool SqlAllChunks::Open() { return true; }


bool SqlAllChunks::Next(shash::Any *hash, zlib::Algorithms *compression_alg) {
  if (!FetchRow()) {
    return false;
  }

  *hash = RetrieveHashBlob(0, static_cast<shash::Algorithms>(RetrieveInt(2)),
                           static_cast<shash::Suffix>(RetrieveInt(1)));
  *compression_alg = static_cast<zlib::Algorithms>(RetrieveInt(3));
  return true;
}


bool SqlAllChunks::Close() { return Reset(); }


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


SqlLookupXattrs::SqlLookupXattrs(const CatalogDatabase &database) {
  DeferredInit(database.sqlite_db(),
               "SELECT xattr FROM catalog "
               "WHERE (md5path_1 = :md5_1) AND (md5path_2 = :md5_2);");
}


bool SqlLookupXattrs::BindPathHash(const shash::Md5 &hash) {
  return BindMd5(1, 2, hash);
}


XattrList SqlLookupXattrs::GetXattrs() {
  const unsigned char *packed_xattrs = reinterpret_cast<const unsigned char *>(
      RetrieveBlob(0));
  if (packed_xattrs == NULL)
    return XattrList();

  const int size = RetrieveBytes(0);
  assert(size >= 0);
  const UniquePtr<XattrList> xattrs(
      XattrList::Deserialize(packed_xattrs, size));
  if (!xattrs.IsValid()) {
    LogCvmfs(kLogCatalog, kLogDebug, "corrupted xattr data");
    return XattrList();
  }
  return *xattrs;
}

}  // namespace catalog