YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

#include "yb/master/catalog_loaders.h"

#include "yb/master/master_util.h"

#include "yb/master/ysql_transaction_ddl.h"

#include "yb/util/status_format.h"

#include "yb/util/status_log.h"

DEFINE_bool(master_ignore_deleted_on_load, true,

  "Whether the Master should ignore deleted tables & tablets on restart.  "

  "This reduces failover time at the expense of garbage data." );

namespace yb {

namespace master {

using namespace std::placeholders;

////////////////////////////////////////////////////////////

// Table Loader

////////////////////////////////////////////////////////////

bool ShouldLoadObject(const SysTablesEntryPB& metadata) {

  // TODO: We need to properly remove deleted tables.  This can happen async of master loading.

  return !FLAGS_master_ignore_deleted_on_load || metadata.state() != SysTablesEntryPB::DELETED;

}

Status TableLoader::Visit(const TableId& table_id, const SysTablesEntryPB& metadata) {

  // TODO: We need to properly remove deleted tables.  This can happen async of master loading.

  if (!ShouldLoadObject(metadata)) {

    return Status::OK();

  }

  CHECK(!ContainsKey(*catalog_manager_->table_ids_map_, table_id))

        << "Table already exists: " << table_id;

  // Setup the table info.

  scoped_refptr<TableInfo> table = catalog_manager_->NewTableInfo(table_id);

  auto l = table->LockForWrite();

  auto& pb = l.mutable_data()->pb;

  pb.CopyFrom(metadata);

  if (pb.table_type() == TableType::REDIS_TABLE_TYPE && pb.name() == kGlobalTransactionsTableName) {

    pb.set_table_type(TableType::TRANSACTION_STATUS_TABLE_TYPE);

  }

  // Add the table to the IDs map and to the name map (if the table is not deleted). Do not

  // add Postgres tables to the name map as the table name is not unique in a namespace.

  auto table_ids_map_checkout = catalog_manager_->table_ids_map_.CheckOut();

  (*table_ids_map_checkout)[table->id()] = table;

  if (!l->started_deleting() && !l->started_hiding()) {

    if (l->table_type() != PGSQL_TABLE_TYPE) {

      catalog_manager_->table_names_map_[{l->namespace_id(), l->name()}] = table;

    }

    if (l->table_type() == TRANSACTION_STATUS_TABLE_TYPE) {

      catalog_manager_->transaction_table_ids_set_.insert(table_id);

    }

  }

  l.Commit();

  catalog_manager_->HandleNewTableId(table->id());

  // Tables created as part of a Transaction should check transaction status and be deleted

  // if the transaction is aborted.

  if (metadata.has_transaction()) {

    LOG(INFO) << "Enqueuing table for Transaction Verification: " << table->ToString();

    TransactionMetadata txn = VERIFY_RESULT(TransactionMetadata::FromPB(metadata.transaction()));

    std::function<Status(bool)> when_done =

        std::bind(&CatalogManager::VerifyTablePgLayer, catalog_manager_, table, _1);

    WARN_NOT_OK(catalog_manager_->background_tasks_thread_pool_->SubmitFunc(

        std::bind(&YsqlTransactionDdl::VerifyTransaction, catalog_manager_->ysql_transaction_.get(),

                  txn, when_done)),

        "Could not submit VerifyTransaction to thread pool");

  }

  LOG(INFO) << "Loaded metadata for table " << table->ToString() << ", state: "

            << SysTablesEntryPB::State_Name(metadata.state());

  VLOG(1) << "Metadata for table " << table->ToString() << ": " << metadata.ShortDebugString();

  return Status::OK();

}

////////////////////////////////////////////////////////////

// Tablet Loader

////////////////////////////////////////////////////////////

bool ShouldLoadObject(const SysTabletsEntryPB& pb) {

  return true;

}

Status TabletLoader::Visit(const TabletId& tablet_id, const SysTabletsEntryPB& metadata) {

  if (!ShouldLoadObject(metadata)) {

    return Status::OK();

  }

  // Lookup the table.

  TableInfoPtr first_table = FindPtrOrNull(*catalog_manager_->table_ids_map_, metadata.table_id());

  // TODO: We need to properly remove deleted tablets.  This can happen async of master loading.

  if (!first_table) {

    if (metadata.state() != SysTabletsEntryPB::DELETED) {

      LOG(DFATAL) << "Unexpected Tablet state for " << tablet_id << ": "

                  << SysTabletsEntryPB::State_Name(metadata.state())

                  << ", unknown table for this tablet: " << metadata.table_id();

    }

    return Status::OK();

  }

  // Setup the tablet info.

  std::vector<TableId> table_ids;

  bool tablet_deleted;

  bool listed_as_hidden;

  TabletInfoPtr tablet(new TabletInfo(first_table, tablet_id));

  {

    auto l = tablet->LockForWrite();

    l.mutable_data()->pb.CopyFrom(metadata);

    // Add the tablet to the tablet manager.

    auto tablet_map_checkout = catalog_manager_->tablet_map_.CheckOut();

    auto inserted = tablet_map_checkout->emplace(tablet->tablet_id(), tablet).second;

    if (!inserted) {

      return STATUS_FORMAT(

          IllegalState, "Loaded tablet that already in map: $0", tablet->tablet_id());

    }

    for (int k = 0; k < metadata.table_ids_size(); ++k) {

      table_ids.push_back(metadata.table_ids(k));

    }

    // This is for backwards compatibility: we want to ensure that the table_ids

    // list contains the first table that created the tablet. If the table_ids field

    // was empty, we "upgrade" the master to support this new invariant.

    if (metadata.table_ids_size() == 0) {

      l.mutable_data()->pb.add_table_ids(metadata.table_id());

      Status s = catalog_manager_->sys_catalog_->Upsert(

          catalog_manager_->leader_ready_term(), tablet);

      if (PREDICT_FALSE(!s.ok())) {

        return STATUS_FORMAT(

            IllegalState, "An error occurred while inserting to sys-tablets: $0", s);

      }

      table_ids.push_back(metadata.table_id());

    }

    tablet_deleted = l.mutable_data()->is_deleted();

    listed_as_hidden = l.mutable_data()->ListedAsHidden();

    // Assume we need to delete this tablet until we find an active table using this tablet.

    bool should_delete_tablet = !tablet_deleted;

    for (const auto& table_id : table_ids) {

      TableInfoPtr table = FindPtrOrNull(*catalog_manager_->table_ids_map_, table_id);

      if (table == nullptr) {

        // If the table is missing and the tablet is in "preparing" state

        // may mean that the table was not created (maybe due to a failed write

        // for the sys-tablets). The cleaner will remove.

        auto tablet_state = l->pb.state();

        if (tablet_state == SysTabletsEntryPB::PREPARING) {

          LOG(WARNING) << "Missing table " << table_id << " required by tablet " << tablet_id

                        << " (probably a failed table creation: the tablet was not assigned)";

          return Status::OK();

        }

        // Otherwise, something is wrong...

        LOG(WARNING) << "Missing table " << table_id << " required by tablet " << tablet_id

                     << ", metadata: " << metadata.DebugString()

                     << ", tables: " << yb::ToString(*catalog_manager_->table_ids_map_);

        // If we ignore deleted tables, then a missing table can be expected and we continue.

        if (PREDICT_TRUE(FLAGS_master_ignore_deleted_on_load)) {

          continue;

        }

        // Otherwise, we need to surface the corruption.

        return STATUS(Corruption, "Missing table for tablet: ", tablet_id);

      }

      // Add the tablet to the Table.

      if (!tablet_deleted) {

        // Any table listed under the sys catalog tablet, is by definition a system table.

        // This is the easiest place to mark these as system tables, as we'll only go over

        // sys_catalog tablet once and can mark in memory all the relevant tables.

        if (tablet_id == kSysCatalogTabletId) {

          table->set_is_system();

        }

        table->AddTablet(tablet.get());

      }

      auto tl = table->LockForRead();

      if (!tl->started_deleting()) {

        // Found an active table.

        should_delete_tablet = false;

      }

    }

    if (should_delete_tablet) {

      LOG(WARNING)

          << "Deleting tablet " << tablet->id() << " for table " << first_table->ToString();

      string deletion_msg = "Tablet deleted at " + LocalTimeAsString();

      l.mutable_data()->set_state(SysTabletsEntryPB::DELETED, deletion_msg);

      RETURN_NOT_OK_PREPEND(catalog_manager_->sys_catalog()->Upsert(term_, tablet),

                            Format("Error deleting tablet $0", tablet->id()));

    }

    l.Commit();

  }

  // Add the tablet to colocated_tablet_ids_map_ if the tablet is colocated.

  if (first_table->IsColocatedParentTable()) {

    catalog_manager_->colocated_tablet_ids_map_[first_table->namespace_id()] =

        catalog_manager_->tablet_map_->find(tablet_id)->second;

  }

  // Add the tablet to tablegroup_tablet_ids_map_ if the tablet is a tablegroup parent.

  if (first_table->IsTablegroupParentTable()) {

    catalog_manager_->tablegroup_tablet_ids_map_[first_table->namespace_id()]

        [first_table->id().substr(0, 32)] = catalog_manager_->tablet_map_->find(tablet_id)->second;

    TablegroupInfo *tg = new TablegroupInfo(first_table->id().substr(0, 32),

                                            first_table->namespace_id());

    // Loop through table_ids again to add them to our tablegroup info.

    for (auto table_id : table_ids) {

      tg->AddChildTable(table_id);

    }

    catalog_manager_->tablegroup_ids_map_[first_table->id().substr(0, 32)] = tg;

  }

  LOG(INFO) << "Loaded metadata for " << (tablet_deleted ? "deleted " : "")

            << "tablet " << tablet_id

            << " (first table " << first_table->ToString() << ")";

  VLOG(1) << "Metadata for tablet " << tablet_id << ": " << metadata.ShortDebugString();

  if (listed_as_hidden) {

    catalog_manager_->hidden_tablets_.push_back(tablet);

  }

  return Status::OK();

}

////////////////////////////////////////////////////////////

// Namespace Loader

////////////////////////////////////////////////////////////

bool ShouldLoadObject(const SysNamespaceEntryPB& metadata) {

  return true;

}

Status NamespaceLoader::Visit(const NamespaceId& ns_id, const SysNamespaceEntryPB& metadata) {

  if (!ShouldLoadObject(metadata)) {

    return Status::OK();

  }

  CHECK(!ContainsKey(catalog_manager_->namespace_ids_map_, ns_id))

    << "Namespace already exists: " << ns_id;

  // Setup the namespace info.

  scoped_refptr<NamespaceInfo> ns = new NamespaceInfo(ns_id);

  auto l = ns->LockForWrite();

  const auto& pb_data = l->pb;

  l.mutable_data()->pb.CopyFrom(metadata);

  if (!pb_data.has_database_type() || pb_data.database_type() == YQL_DATABASE_UNKNOWN) {

    LOG(INFO) << "Updating database type of namespace " << pb_data.name();

    l.mutable_data()->pb.set_database_type(GetDefaultDatabaseType(pb_data.name()));

  }

  // When upgrading, we won't have persisted this new field.

  // TODO: Persist this change to disk instead of just changing memory.

  auto state = metadata.state();

  if (!metadata.has_state()) {

    state = SysNamespaceEntryPB::RUNNING;

    LOG(INFO) << "Changing metadata without state to RUNNING: " << ns->ToString();

    l.mutable_data()->pb.set_state(state);

  }

  switch(state) {

    case SysNamespaceEntryPB::RUNNING:

      // Add the namespace to the IDs map and to the name map (if the namespace is not deleted).

      catalog_manager_->namespace_ids_map_[ns_id] = ns;

      if (!pb_data.name().empty()) {

        catalog_manager_->namespace_names_mapper_[pb_data.database_type()][pb_data.name()] = ns;

      } else {

        LOG(WARNING) << "Namespace with id " << ns_id << " has empty name";

      }

      l.Commit();

      LOG(INFO) << "Loaded metadata for namespace " << ns->ToString();

      // Namespaces created as part of a Transaction should check transaction status and be deleted

      // if the transaction is aborted.

      if (metadata.has_transaction()) {

        LOG(INFO) << "Enqueuing keyspace for Transaction Verification: " << ns->ToString();

        TransactionMetadata txn = VERIFY_RESULT(

            TransactionMetadata::FromPB(metadata.transaction()));

        std::function<Status(bool)> when_done =

            std::bind(&CatalogManager::VerifyNamespacePgLayer, catalog_manager_, ns, _1);

        WARN_NOT_OK(catalog_manager_->background_tasks_thread_pool_->SubmitFunc(

            std::bind(&YsqlTransactionDdl::VerifyTransaction,

                      catalog_manager_->ysql_transaction_.get(), txn, when_done)),

          "Could not submit VerifyTransaction to thread pool");

      }

      break;

    case SysNamespaceEntryPB::PREPARING:

      // PREPARING means the server restarted before completing NS creation.

      // Consider it FAILED & remove any partially-created data.

      FALLTHROUGH_INTENDED;

    case SysNamespaceEntryPB::FAILED:

      LOG(INFO) << "Transitioning failed namespace (state="  << metadata.state()

                << ") to DELETING: " << ns->ToString();

      l.mutable_data()->pb.set_state(SysNamespaceEntryPB::DELETING);

      FALLTHROUGH_INTENDED;

    case SysNamespaceEntryPB::DELETING:

      catalog_manager_->namespace_ids_map_[ns_id] = ns;

      if (!pb_data.name().empty()) {

        catalog_manager_->namespace_names_mapper_[pb_data.database_type()][pb_data.name()] = ns;

      } else {

        LOG(WARNING) << "Namespace with id " << ns_id << " has empty name";

      }

      l.Commit();

      LOG(INFO) << "Loaded metadata to DELETE namespace " << ns->ToString();

      if (ns->database_type() != YQL_DATABASE_PGSQL) {

        WARN_NOT_OK(catalog_manager_->background_tasks_thread_pool_->SubmitFunc(

          std::bind(&CatalogManager::DeleteYcqlDatabaseAsync, catalog_manager_, ns)),

          "Could not submit DeleteYcqlDatabaseAsync to thread pool");

      } else {

        WARN_NOT_OK(catalog_manager_->background_tasks_thread_pool_->SubmitFunc(

          std::bind(&CatalogManager::DeleteYsqlDatabaseAsync, catalog_manager_, ns)),

          "Could not submit DeleteYsqlDatabaseAsync to thread pool");

      }

      break;

    case SysNamespaceEntryPB::DELETED:

      LOG(INFO) << "Skipping metadata for namespace (state="  << metadata.state()

                << "): " << ns->ToString();

      // Garbage collection.  Async remove the Namespace from the SysCatalog.

      // No in-memory state needed since tablet deletes have already been processed.

      WARN_NOT_OK(catalog_manager_->background_tasks_thread_pool_->SubmitFunc(

          std::bind(&CatalogManager::DeleteYsqlDatabaseAsync, catalog_manager_, ns)),

          "Could not submit DeleteYsqlDatabaseAsync to thread pool");

      break;

    default:

      FATAL_INVALID_ENUM_VALUE(SysNamespaceEntryPB_State, state);

  }

  VLOG(1) << "Metadata for namespace " << ns->ToString() << ": " << metadata.ShortDebugString();

  return Status::OK();

}

////////////////////////////////////////////////////////////

// User-Defined Type Loader

////////////////////////////////////////////////////////////

Status UDTypeLoader::Visit(const UDTypeId& udtype_id, const SysUDTypeEntryPB& metadata) {

  CHECK(!ContainsKey(catalog_manager_->udtype_ids_map_, udtype_id))

      << "Type already exists: " << udtype_id;

  // Setup the table info.

  UDTypeInfo* const udtype = new UDTypeInfo(udtype_id);

  {

    auto l = udtype->LockForWrite();

    l.mutable_data()->pb.CopyFrom(metadata);

    // Add the used-defined type to the IDs map and to the name map (if the type is not deleted).

    catalog_manager_->udtype_ids_map_[udtype->id()] = udtype;

    if (!l->name().empty()) {  // If name is set (non-empty) then type is not deleted.

      catalog_manager_->udtype_names_map_[{l->namespace_id(), l->name()}] = udtype;

    }

    l.Commit();

  }

  LOG(INFO) << "Loaded metadata for type " << udtype->ToString();

  VLOG(1) << "Metadata for type " << udtype->ToString() << ": " << metadata.ShortDebugString();

  return Status::OK();

}

////////////////////////////////////////////////////////////

// Config Loader

////////////////////////////////////////////////////////////

Status ClusterConfigLoader::Visit(

    const std::string& unused_id, const SysClusterConfigEntryPB& metadata) {

  // Debug confirm that there is no cluster_config_ set. This also ensures that this does not

  // visit multiple rows. Should update this, if we decide to have multiple IDs set as well.

  DCHECK(!catalog_manager_->cluster_config_) << "Already have config data!";

  // Prepare the config object.

  ClusterConfigInfo* config = new ClusterConfigInfo();

  {

    auto l = config->LockForWrite();

    l.mutable_data()->pb.CopyFrom(metadata);

    // Update in memory state.

    catalog_manager_->cluster_config_ = config;

    l.Commit();

  }

  return Status::OK();

}

////////////////////////////////////////////////////////////

// Redis Config Loader

////////////////////////////////////////////////////////////

Status RedisConfigLoader::Visit(const std::string& key, const SysRedisConfigEntryPB& metadata) {

  CHECK(!ContainsKey(catalog_manager_->redis_config_map_, key))

      << "Redis Config with key already exists: " << key;

  // Prepare the config object.

  RedisConfigInfo* config = new RedisConfigInfo(key);

  {

    auto l = config->LockForWrite();

    l.mutable_data()->pb.CopyFrom(metadata);

    catalog_manager_->redis_config_map_[key] = config;

    l.Commit();

  }

  return Status::OK();

}

////////////////////////////////////////////////////////////

// Role Loader

////////////////////////////////////////////////////////////

Status RoleLoader::Visit(const RoleName& role_name, const SysRoleEntryPB& metadata) {

  RoleInfo* const role = new RoleInfo(role_name);

  {

    auto l = role->LockForWrite();

    l.mutable_data()->pb.CopyFrom(metadata);

    catalog_manager_->permissions_manager()->AddRoleUnlocked(

        role_name, make_scoped_refptr<RoleInfo>(role));

    l.Commit();

  }

  LOG(INFO) << "Loaded metadata for role " << role->id();

  VLOG(1) << "Metadata for role " << role->id() << ": " << metadata.ShortDebugString();

  return Status::OK();

}

////////////////////////////////////////////////////////////

// Sys Config Loader

////////////////////////////////////////////////////////////

Status SysConfigLoader::Visit(const string& config_type, const SysConfigEntryPB& metadata) {

  SysConfigInfo* const config = new SysConfigInfo(config_type);

  {

    auto l = config->LockForWrite();

    l.mutable_data()->pb.CopyFrom(metadata);

    if (config_type == kSecurityConfigType) {

      catalog_manager_->permissions_manager()->SetSecurityConfigOnLoadUnlocked(config);

    } else if (config_type == kYsqlCatalogConfigType) {

      LOG_IF(WARNING, catalog_manager_->ysql_catalog_config_ != nullptr)

          << "Multiple sys config type " << config_type << " found";

      catalog_manager_->ysql_catalog_config_ = config;

    } else if (config_type == kTransactionTablesConfigType) {

      LOG_IF(WARNING, catalog_manager_->transaction_tables_config_ != nullptr)

          << "Multiple sys config type " << config_type << " found";

      catalog_manager_->transaction_tables_config_ = config;

    }

    l.Commit();

  }

  LOG(INFO) << "Loaded sys config type " << config_type;

  return Status::OK();

}

}  // namespace master

}  // namespace yb