YugabyteDB (2.13.1.0-b60, 21121d69985fbf76aa6958d8f04a9bfa936293b5)

// 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/backfill_index.h"

#include <pthread.h>

#include <stdlib.h>

#include <sys/types.h>

#include <algorithm>

#include <bitset>

#include <functional>

#include <memory>

#include <mutex>

#include <set>

#include <string>

#include <unordered_map>

#include <vector>

#include <boost/optional.hpp>

#include <boost/preprocessor/cat.hpp>

#include <glog/logging.h>

#include "yb/common/partial_row.h"

#include "yb/common/partition.h"

#include "yb/common/wire_protocol.h"

#include "yb/docdb/doc_rowwise_iterator.h"

#include "yb/gutil/atomicops.h"

#include "yb/gutil/callback.h"

#include "yb/gutil/casts.h"

#include "yb/gutil/map-util.h"

#include "yb/gutil/mathlimits.h"

#include "yb/gutil/ref_counted.h"

#include "yb/gutil/stl_util.h"

#include "yb/gutil/strings/escaping.h"

#include "yb/gutil/strings/join.h"

#include "yb/gutil/strings/substitute.h"

#include "yb/gutil/sysinfo.h"

#include "yb/master/master_fwd.h"

#include "yb/master/async_rpc_tasks.h"

#include "yb/master/catalog_manager.h"

#include "yb/master/master.h"

#include "yb/master/master_ddl.pb.h"

#include "yb/master/sys_catalog.h"

#include "yb/tablet/tablet.h"

#include "yb/tablet/tablet_metadata.h"

#include "yb/tablet/tablet_peer.h"

#include "yb/tserver/tserver_admin.proxy.h"

#include "yb/util/flag_tags.h"

#include "yb/util/format.h"

#include "yb/util/math_util.h"

#include "yb/util/monotime.h"

#include "yb/util/random_util.h"

#include "yb/util/result.h"

#include "yb/util/status_format.h"

#include "yb/util/status_log.h"

#include "yb/util/threadpool.h"

#include "yb/util/trace.h"

#include "yb/util/uuid.h"

DEFINE_int32(ysql_index_backfill_rpc_timeout_ms, 60 * 1000, // 1 min.

             "Timeout used by the master when attempting to backfill a YSQL tablet during index "

             "creation.");

TAG_FLAG(ysql_index_backfill_rpc_timeout_ms, advanced);

TAG_FLAG(ysql_index_backfill_rpc_timeout_ms, runtime);

DEFINE_int32(index_backfill_rpc_timeout_ms, 1 * 30 * 1000, // 30 sec.

             "Timeout used by the master when attempting to backfill a tablet "

             "during index creation.");

TAG_FLAG(index_backfill_rpc_timeout_ms, advanced);

TAG_FLAG(index_backfill_rpc_timeout_ms, runtime);

DEFINE_int32(index_backfill_rpc_max_retries, 150,

             "Number of times to retry backfilling a tablet chunk "

             "during index creation.");

TAG_FLAG(index_backfill_rpc_max_retries, advanced);

TAG_FLAG(index_backfill_rpc_max_retries, runtime);

DEFINE_int32(index_backfill_rpc_max_delay_ms, 10 * 60 * 1000, // 10 min.

             "Maximum delay before retrying a backfill tablet chunk request "

             "during index creation.");

TAG_FLAG(index_backfill_rpc_max_delay_ms, advanced);

TAG_FLAG(index_backfill_rpc_max_delay_ms, runtime);

DEFINE_int32(index_backfill_wait_for_alter_table_completion_ms, 100,

             "Delay before retrying to see if an in-progress alter table has "

             "completed, during index backfill.");

TAG_FLAG(index_backfill_wait_for_alter_table_completion_ms, advanced);

TAG_FLAG(index_backfill_wait_for_alter_table_completion_ms, runtime);

DEFINE_bool(defer_index_backfill, false,

            "Defer index backfill so that backfills can be performed as a batch later on.");

TAG_FLAG(defer_index_backfill, advanced);

TAG_FLAG(defer_index_backfill, runtime);

DEFINE_test_flag(int32, slowdown_backfill_alter_table_rpcs_ms, 0,

    "Slows down the send alter table rpc's so that the master may be stopped between "

    "different phases.");

DEFINE_test_flag(

    int32, slowdown_backfill_job_deletion_ms, 0,

    "Slows down backfill job deletion so that backfill job can be read by test.");

namespace yb {

namespace master {

using namespace std::literals;

using server::MonitoredTaskState;

using strings::Substitute;

using tserver::TabletServerErrorPB;

namespace {

// Peek into pg_index table to get an index (boolean) status from YSQL perspective.

Result<bool> GetPgIndexStatus(

    CatalogManager* catalog_manager,

    const TableId& idx_id,

    const std::string& status_col_name) {

  const auto pg_index_id =

      GetPgsqlTableId(VERIFY_RESULT(GetPgsqlDatabaseOid(idx_id)), kPgIndexTableOid);

  const tablet::Tablet* catalog_tablet =

      catalog_manager->tablet_peer()->tablet();

  const Schema& pg_index_schema =

      *VERIFY_RESULT(catalog_tablet->metadata()->GetTableInfo(pg_index_id))->schema;

  Schema projection;

  RETURN_NOT_OK(pg_index_schema.CreateProjectionByNames({"indexrelid", status_col_name},

                                                        &projection,

                                                        pg_index_schema.num_key_columns()));

  const auto indexrelid_col_id = VERIFY_RESULT(projection.ColumnIdByName("indexrelid")).rep();

  const auto status_col_id     = VERIFY_RESULT(projection.ColumnIdByName(status_col_name)).rep();

  const auto idx_oid = VERIFY_RESULT(GetPgsqlTableOid(idx_id));

  auto iter = VERIFY_RESULT(catalog_tablet->NewRowIterator(projection.CopyWithoutColumnIds(),

                                                           {} /* read_hybrid_time */,

                                                           pg_index_id));

  // Filtering by 'indexrelid' == idx_oid.

  {

    auto doc_iter = down_cast<docdb::DocRowwiseIterator*>(iter.get());

    PgsqlConditionPB cond;

    cond.add_operands()->set_column_id(indexrelid_col_id);

    cond.set_op(QL_OP_EQUAL);

    cond.add_operands()->mutable_value()->set_uint32_value(idx_oid);

    const std::vector<docdb::PrimitiveValue> empty_key_components;

    docdb::DocPgsqlScanSpec spec(projection,

                                 rocksdb::kDefaultQueryId,

                                 empty_key_components,

                                 empty_key_components,

                                 &cond,

                                 boost::none /* hash_code */,

                                 boost::none /* max_hash_code */,

                                 nullptr /* where_expr */);

    RETURN_NOT_OK(doc_iter->Init(spec));

  }

  // Expecting one row at most.

  QLTableRow row;

  if (VERIFY_RESULT(iter->HasNext())) {

    RETURN_NOT_OK(iter->NextRow(&row));

    return row.GetColumn(status_col_id)->bool_value();

  }

  // For practical purposes, an absent index is the same as having false status column value.

  return false;

}

// Before advancing index permissions, we need to make sure Postgres side has advanced sufficiently

// - that the state tracked in pg_index haven't fallen behind from the desired permission

// for more than one step.

Result<bool> ShouldProceedWithPgsqlIndexPermissionUpdate(

    CatalogManager* catalog_manager,

    const TableId& idx_id,

    IndexPermissions new_perm) {

  // TODO(alex, jason): Add the appropriate cases for dropping index path

  switch (new_perm) {

    case INDEX_PERM_WRITE_AND_DELETE: {

      auto live = VERIFY_RESULT(GetPgIndexStatus(catalog_manager, idx_id, "indislive"));

      if (!live) {

        VLOG(1) << "Index " << idx_id << " is not yet live, skipping permission update";

      }

      return live;

    }

    case INDEX_PERM_DO_BACKFILL: {

      auto ready = VERIFY_RESULT(GetPgIndexStatus(catalog_manager, idx_id, "indisready"));

      if (!ready) {

        VLOG(1) << "Index " << idx_id << " is not yet ready, skipping permission update";

      }

      return ready;

    }

    default:

      // No need to wait for anything

      return true;

  }

}

} // namespace

void MultiStageAlterTable::CopySchemaDetailsToFullyApplied(SysTablesEntryPB* pb) {

  VLOG
(4) << "Setting fully_applied_schema_version to " << pb->version()0
;

  pb->mutable_fully_applied_schema()->CopyFrom(pb->schema());

  pb->set_fully_applied_schema_version(pb->version());

  pb->mutable_fully_applied_indexes()->CopyFrom(pb->indexes());

  if (pb->has_index_info()) {

    pb->mutable_fully_applied_index_info()->CopyFrom(pb->index_info());

  }

}

Status MultiStageAlterTable::ClearFullyAppliedAndUpdateState(

    CatalogManager* catalog_manager,

    const scoped_refptr<TableInfo>& table,

    boost::optional<uint32_t> expected_version,

    bool update_state_to_running) {

  auto l = table->LockForWrite();

  uint32_t current_version = l->pb.version();

  if (expected_version && 
*expected_version != current_version9.30k
) {

    return STATUS(AlreadyPresent, "Table has already moved to a different version.");

  }

  l.mutable_data()->pb.clear_fully_applied_schema();

  l.mutable_data()->pb.clear_fully_applied_schema_version();

  l.mutable_data()->pb.clear_fully_applied_indexes();

  l.mutable_data()->pb.clear_fully_applied_index_info();

  auto new_state = update_state_to_running ? 
SysTablesEntryPB::RUNNING8.89k
 : 
SysTablesEntryPB::ALTERING950
;

  l.mutable_data()->set_state(new_state, Format("Current schema version=$0", current_version));

  Status s = catalog_manager->sys_catalog_->Upsert(catalog_manager->leader_ready_term(), table);

  if (!s.ok()) {

    LOG(WARNING) << "An error occurred while updating sys-tables: " << s.ToString()

                 << ". This master may not be the leader anymore.";

    return s;

  }

  l.Commit();

  LOG(INFO) << table->ToString() << " - Alter table completed version=" << current_version

            << ", state: " << SysTablesEntryPB::State_Name(new_state);

  return Status::OK();

}

Result<bool> MultiStageAlterTable::UpdateIndexPermission(

    CatalogManager* catalog_manager,

    const scoped_refptr<TableInfo>& indexed_table,

    const std::unordered_map<TableId, IndexPermissions>& perm_mapping,

    boost::optional<uint32_t> current_version) {

  TRACE(__func__);

  DVLOG
(3) << __PRETTY_FUNCTION__ << " " << yb::ToString(*indexed_table)5
;

  if (FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms > 0) {

    TRACE("Sleeping for $0 ms", FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms);

    DVLOG(3) << __PRETTY_FUNCTION__ << " " << yb::ToString(*indexed_table) << " sleeping for "

             << FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms

             << "ms BEFORE updating the index permission to " << ToString(perm_mapping);

    SleepFor(MonoDelta::FromMilliseconds(FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms));

    DVLOG
(3) << __PRETTY_FUNCTION__ << " Done Sleeping"7
;

    TRACE("Done Sleeping");

  }

  bool permissions_updated = false;

  {

    TRACE("Locking indexed table");

    auto l = indexed_table->LockForWrite();

    auto& indexed_table_data = *l.mutable_data();

    auto& indexed_table_pb = indexed_table_data.pb;

    if (current_version && 
*current_version != indexed_table_pb.version()1.14k
) {

      LOG(INFO) << "The table schema version "

                << "seems to have already been updated to " << indexed_table_pb.version()

                << " We wanted to do this update at " << *current_version;

      return STATUS_SUBSTITUTE(

          AlreadyPresent, "Schema was already updated to $0 before we got to it (expected $1).",

          indexed_table_pb.version(), *current_version);

    }

    CopySchemaDetailsToFullyApplied(&indexed_table_pb);

    bool is_pgsql = indexed_table_pb.table_type() == TableType::PGSQL_TABLE_TYPE;

    for (int i = 0; i < indexed_table_pb.indexes_size(); 
i++3.52k
) {

      IndexInfoPB* idx_pb = indexed_table_pb.mutable_indexes(i);

      auto& idx_table_id = idx_pb->table_id();

      if (perm_mapping.find(idx_table_id) != perm_mapping.end()) {

        const auto new_perm = perm_mapping.at(idx_table_id);

        if (idx_pb->index_permissions() >= new_perm) {

          LOG(WARNING) << "Index " << idx_pb->table_id() << " on table "

                       << indexed_table->ToString() << " has index_permission "

                       << IndexPermissions_Name(idx_pb->index_permissions()) << " already past "

                       << IndexPermissions_Name(new_perm) << ". Will not update it";

          continue;

        }

        // TODO(alex, amit): Non-OK status here should be converted to TryAgain,

        //                   which should be handled on an upper level.

        if (is_pgsql && 
!534
VERIFY_RESULT534
(ShouldProceedWithPgsqlIndexPermissionUpdate(catalog_manager,

                                                                                   idx_table_id,

                                                                                   new_perm))) {

          continue;

        }

        idx_pb->set_index_permissions(new_perm);

        permissions_updated = true;

      }

    }

    if (permissions_updated) {

      indexed_table_pb.set_version(indexed_table_pb.version() + 1);

      indexed_table_pb.set_updates_only_index_permissions(true);

    } else {

      VLOG(1) << "Index permissions update skipped, leaving schema_version at "

              << indexed_table_pb.version();

    }

    indexed_table_data.set_state(

        SysTablesEntryPB::ALTERING,

        Format("Update index permission version=$0 ts=$1",

               indexed_table_pb.version(), LocalTimeAsString()));

    // Update sys-catalog with the new indexed table info.

    TRACE("Updating indexed table metadata on disk");

    RETURN_NOT_OK(catalog_manager->sys_catalog_->Upsert(

        catalog_manager->leader_ready_term(), indexed_table));

    // Update the in-memory state.

    TRACE("Committing in-memory state");

    l.Commit();

  }

  if (PREDICT_FALSE(FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms > 0)) {

    TRACE("Sleeping for $0 ms",

          FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms);

    DVLOG(3) << __PRETTY_FUNCTION__ << " " << yb::ToString(*indexed_table) << " sleeping for "

             << FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms

             << "ms AFTER updating the index permission to " << ToString(perm_mapping);

    SleepFor(MonoDelta::FromMilliseconds(FLAGS_TEST_slowdown_backfill_alter_table_rpcs_ms));

    DVLOG
(3) << __PRETTY_FUNCTION__ << " Done Sleeping"13
;

    TRACE("Done Sleeping");

  }

  return permissions_updated;

}

Status MultiStageAlterTable::StartBackfillingData(

    CatalogManager* catalog_manager,

    const scoped_refptr<TableInfo>& indexed_table,

    const std::vector<IndexInfoPB>& idx_infos,

    boost::optional<uint32_t> current_version) {

  // We leave the table state as ALTERING so that a master failover can resume the backfill.

  RETURN_NOT_OK(ClearFullyAppliedAndUpdateState(

      catalog_manager, indexed_table, current_version, /* change_state to RUNNING */ false));

  RETURN_NOT_OK(indexed_table->SetIsBackfilling());

  TRACE("Starting backfill process");

  VLOG(0) << __func__ << " starting backfill on " << indexed_table->ToString() << " for "

          << yb::ToString(idx_infos);

  auto ns_info = catalog_manager->FindNamespaceById(indexed_table->namespace_id());

  RETURN_NOT_OK_PREPEND(ns_info, "Unable to get namespace info for backfill");

  auto backfill_table = std::make_shared<BackfillTable>(

      catalog_manager->master_, catalog_manager->AsyncTaskPool(), indexed_table, idx_infos,

      *ns_info);

  backfill_table->Launch();

  return Status::OK();

}

// Returns true, if the said IndexPermissions is a transient state.

// Returns false, if it is a state where the index can be. viz: READ_WRITE_AND_DELETE

// INDEX_UNUSED is considered transcient because it needs to delete the index.

bool IsTransientState(IndexPermissions perm) {

  return perm != INDEX_PERM_READ_WRITE_AND_DELETE && perm != INDEX_PERM_NOT_USED;

}

IndexPermissions NextPermission(IndexPermissions perm) {

  switch (perm) {

    case INDEX_PERM_DELETE_ONLY:

      return INDEX_PERM_WRITE_AND_DELETE;

    case INDEX_PERM_WRITE_AND_DELETE:

      return INDEX_PERM_DO_BACKFILL;

    case INDEX_PERM_DO_BACKFILL:

      CHECK(false) << "Not expected to be here.";

      return INDEX_PERM_DELETE_ONLY;

    case INDEX_PERM_READ_WRITE_AND_DELETE:

      CHECK(false) << "Not expected to be here.";

      return INDEX_PERM_DELETE_ONLY;

    case INDEX_PERM_WRITE_AND_DELETE_WHILE_REMOVING:

      return INDEX_PERM_DELETE_ONLY_WHILE_REMOVING;

    case INDEX_PERM_DELETE_ONLY_WHILE_REMOVING:

      return INDEX_PERM_INDEX_UNUSED;

    case INDEX_PERM_INDEX_UNUSED:

    case INDEX_PERM_NOT_USED:

      CHECK(false) << "Not expected to be here.";

      return INDEX_PERM_DELETE_ONLY;

  }

  CHECK(false) << "Not expected to be here.";

  return INDEX_PERM_DELETE_ONLY;

}

Status MultiStageAlterTable::LaunchNextTableInfoVersionIfNecessary(

    CatalogManager* catalog_manager, const scoped_refptr<TableInfo>& indexed_table,

    uint32_t current_version, bool respect_backfill_deferrals) {

  DVLOG
(3) << __PRETTY_FUNCTION__ << " " << yb::ToString(*indexed_table)7
;

  const bool is_ysql_table = (indexed_table->GetTableType() == TableType::PGSQL_TABLE_TYPE);

  const bool defer_backfill = !is_ysql_table && 
GetAtomicFlag(&FLAGS_defer_index_backfill)2.47k
;

  const bool is_backfilling = indexed_table->IsBackfilling();

  std::unordered_map<TableId, IndexPermissions> indexes_to_update;

  vector<IndexInfoPB> indexes_to_backfill;

  vector<IndexInfoPB> deferred_indexes;

  vector<IndexInfoPB> indexes_to_delete;

  {

    TRACE("Locking indexed table");

    VLOG
(1) << ("Locking indexed table")9
;

    auto l = indexed_table->LockForRead();

    VLOG
(1) << ("Locked indexed table")11
;

    if (current_version != l->pb.version()) {

      LOG(WARNING) << "Somebody launched the next version before we got to it.";

      return Status::OK();

    }

    // Attempt to find an index that requires us to just launch the next state (i.e. not backfill)

    
for (int i = 0; 10.5k
i < l->pb.indexes_size(); 
i++6.98k
) {

      const IndexInfoPB& idx_pb = l->pb.indexes(i);

      if (!idx_pb.has_index_permissions()) {

        continue;

      }

      if (idx_pb.index_permissions() == INDEX_PERM_DO_BACKFILL) {

        if (respect_backfill_deferrals && 
(639
defer_backfill639
 || 
idx_pb.is_backfill_deferred()639
)) {

          LOG(INFO) << "Deferring index-backfill for " << idx_pb.table_id();

          deferred_indexes.emplace_back(idx_pb);

        } else {

          indexes_to_backfill.emplace_back(idx_pb);

        }

      } else if (idx_pb.index_permissions() == INDEX_PERM_INDEX_UNUSED) {

        indexes_to_delete.emplace_back(idx_pb);

      // For YSQL, there should never be indexes to update from master side because postgres drives

      // permission changes.

      } else if (idx_pb.index_permissions() != INDEX_PERM_READ_WRITE_AND_DELETE && 
!is_ysql_table1.83k
) {

        indexes_to_update.emplace(idx_pb.table_id(), NextPermission(idx_pb.index_permissions()));

      }

    }

    // TODO(#6218): Do we really not want to continue backfill

    // across master failovers for YSQL?

    if (!is_ysql_table && 
!is_backfilling2.47k
 && 
l.data().pb.backfill_jobs_size() > 02.33k
) {

      // If a backfill job was started for a set of indexes and then the leader

      // fails over, we should be careful that we are restarting the backfill job

      // with the same set of indexes.

      // A new index could have been added since the time the last backfill job started on

      // the old master. The safe time calculated for the earlier set of indexes may not be

      // valid for the new index(es) to use.

      DCHECK(l.data().pb.backfill_jobs_size() == 1) << "For now we only expect to have up to 1 "

                                                        "outstanding backfill job.";

      const BackfillJobPB& backfill_job = l.data().pb.backfill_jobs(0);

      VLOG(3) << "Found an in-progress backfill-job " << yb::ToString(backfill_job);

      // Do not allow for any other indexes to piggy back with this backfill.

      indexes_to_backfill.clear();

      deferred_indexes.clear();

      for (int i = 0; i < backfill_job.indexes_size(); i++) {

        const IndexInfoPB& idx_pb = backfill_job.indexes(i);

        indexes_to_backfill.push_back(idx_pb);

      }

    }

  }

  if (indexes_to_update.empty() &&

      
indexes_to_delete.empty()9.43k
 &&

      
(9.31k
is_backfilling9.31k
 || 
indexes_to_backfill.empty()9.27k
)) {

    TRACE("Not necessary to launch next version");

    VLOG
(1) << "Not necessary to launch next version"4
;

    return ClearFullyAppliedAndUpdateState(

        catalog_manager, indexed_table, current_version, /* change state to RUNNING */ true);

  }

  // For YSQL online schema migration of indexes, instead of master driving the schema changes,

  // postgres will drive it.  Postgres will use three of the DocDB index permissions:

  //

  // - INDEX_PERM_WRITE_AND_DELETE (set from the start)

  // - INDEX_PERM_READ_WRITE_AND_DELETE (set by master)

  // - INDEX_PERM_WRITE_AND_DELETE_WHILE_REMOVING (set by master)

  //

  // This changes how we treat indexes_to_foo:

  //

  // - indexes_to_update should always be empty because we never want master to set index

  //   permissions.

  // - indexes_to_delete is impossible to be nonempty, and, in the future, when we do use

  //   INDEX_PERM_INDEX_UNUSED, we want to use some other delete trigger that makes sure no

  //   transactions are left using the index.  Prepare for that by doing nothing when nonempty.

  // - indexes_to_backfill is impossible to be nonempty, but, in the future, we want to set

  //   INDEX_PERM_DO_BACKFILL so that backfill resumes on master leader changes.  Prepare for that

  //   by handling indexes_to_backfill like for YCQL.

  //

  // TODO(jason): when using INDEX_PERM_DO_BACKFILL, update this comment (issue #6218).

  if (!indexes_to_update.empty()) {

    VLOG(1) << "Updating index permissions for " << yb::ToString(indexes_to_update) << " on "

            << indexed_table->ToString();

    Result<bool> permissions_updated =

        
VERIFY_RESULT966
(UpdateIndexPermission(catalog_manager, indexed_table, indexes_to_update,

                                            current_version));

    if (!permissions_updated.ok()) {

      LOG(WARNING) << "Could not update index permissions."

                   << " Possible that the master-leader has changed, or a race "

                   << "with another thread trying to launch next version: "

                   << permissions_updated.ToString();

    }

    if (permissions_updated.ok() && *permissions_updated) {

      VLOG
(1) << "Sending alter table request with updated permissions"4
;

      RETURN_NOT_OK(catalog_manager->SendAlterTableRequest(indexed_table));

      return Status::OK();

    }

  }

  if (!indexes_to_delete.empty()) {

    const auto& index_info_to_update = indexes_to_delete[0];

    VLOG(3) << "Deleting the index and the entry in the indexed table for "

            << yb::ToString(index_info_to_update);

    DeleteTableRequestPB req;

    DeleteTableResponsePB resp;

    req.mutable_table()->set_table_id(index_info_to_update.table_id());

    req.set_is_index_table(true);

    RETURN_NOT_OK(catalog_manager->DeleteTableInternal(&req, &resp, nullptr));

    return Status::OK();

  }

  
if (406
!indexes_to_backfill.empty()406
) {

    VLOG(3) << "Backfilling " << yb::ToString(indexes_to_backfill)

            << (deferred_indexes.empty()

                 ? ""

                 : yb::Format(" along with deferred indexes $0",

                              yb::ToString(deferred_indexes)));

    for (auto& deferred_idx : deferred_indexes) {

      indexes_to_backfill.emplace_back(deferred_idx);

    }

    WARN_NOT_OK(

        StartBackfillingData(

            catalog_manager, indexed_table.get(), indexes_to_backfill, current_version),

        yb::Format("Could not launch backfill for $0", indexed_table->ToString()));

  }

  return Status::OK();

}

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

// BackfillTableJob

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

std::string BackfillTableJob::description() const {

  const std::shared_ptr<BackfillTable> retain_bt = backfill_table_;

  auto curr_state = state();

  if (!IsStateTerminal(curr_state) && retain_bt) {

    return retain_bt->description();

  } else if (curr_state == MonitoredTaskState::kFailed) {

    return Format("Backfilling $0 Failed", requested_index_names_);

  } else if (curr_state == MonitoredTaskState::kAborted) {

    return Format("Backfilling $0 Aborted", requested_index_names_);

  } else {

    DCHECK(curr_state == MonitoredTaskState::kComplete);

    return Format("Backfilling $0 Done", requested_index_names_);

  }

}

MonitoredTaskState BackfillTableJob::AbortAndReturnPrevState(const Status& status) {

  auto old_state = state();

  while (!IsStateTerminal(old_state)) {

    if (state_.compare_exchange_strong(old_state,

                                       MonitoredTaskState::kAborted)) {

      return old_state;

    }

    old_state = state();

  }

  return old_state;

}

void BackfillTableJob::SetState(MonitoredTaskState new_state) {

  auto old_state = state();

  if (!IsStateTerminal(old_state)) {

    if (state_.compare_exchange_strong(old_state, new_state) && IsStateTerminal(new_state)) {

      MarkDone();

    }

  }

}

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

// BackfillTable

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

namespace {

std::unordered_set<TableId> IndexIdsFromInfos(const std::vector<IndexInfoPB>& indexes) {

  std::unordered_set<TableId> idx_ids;

  for (const auto& idx_info : indexes) {

    idx_ids.insert(idx_info.table_id());

  }

  return idx_ids;

}

std::string RetrieveIndexNames(CatalogManager* mgr,

                               const std::unordered_set<std::string>& index_ids) {

  std::ostringstream out;

  out << "{ ";

  bool first = true;

  for (const auto& index_id : index_ids) {

    const auto table_info = mgr->GetTableInfo(index_id);

    if (!table_info) {

      LOG(WARNING) << "No table info can be found with index table id " << index_id;

      continue;

    }

    if (!first) {

      out << ", ";

    }

    first = false;

    out << table_info->name();

  }

  out << " }";

  return out.str();

}

}  // namespace

BackfillTable::BackfillTable(

    Master* master, ThreadPool* callback_pool, const scoped_refptr<TableInfo>& indexed_table,

    std::vector<IndexInfoPB> indexes, const scoped_refptr<NamespaceInfo>& ns_info)

    : master_(master),

      callback_pool_(callback_pool),

      indexed_table_(indexed_table),

      index_infos_(indexes),

      requested_index_ids_(IndexIdsFromInfos(indexes)),

      requested_index_names_(RetrieveIndexNames(

          master->catalog_manager_impl(), requested_index_ids_)),

      ns_info_(ns_info) {

  auto l = indexed_table_->LockForRead();

  schema_version_ = indexed_table_->metadata().state().pb.version();

  leader_term_ = master_->catalog_manager()->leader_ready_term();

  if (l.data().pb.backfill_jobs_size() > 0) {

    number_rows_processed_.store(l.data().pb.backfill_jobs(0).num_rows_processed());

  } else {

    number_rows_processed_.store(0);

  }

  const auto& pb = indexed_table_->metadata().state().pb;

  if (pb.backfill_jobs_size() > 0 && 
pb.backfill_jobs(0).has_backfilling_timestamp()0
 &&

      
read_time_for_backfill_.FromUint64(pb.backfill_jobs(0).backfilling_timestamp()).ok()0
) {

    DCHECK(pb.backfill_jobs_size() == 1) << "Expect only 1 outstanding backfill job";

    DCHECK(implicit_cast<size_t>(pb.backfill_jobs(0).indexes_size()) == index_infos_.size())

        << "Expect to use the same set of indexes.";

    timestamp_chosen_.store(true, std::memory_order_release);

    VLOG_WITH_PREFIX(1) << "Will be using " << read_time_for_backfill_

                        << " for backfill";

  } else {

    read_time_for_backfill_ = HybridTime::kInvalid;

    timestamp_chosen_.store(false, std::memory_order_release);

  }

  done_.store(false, std::memory_order_release);

}

const std::unordered_set<TableId> BackfillTable::indexes_to_build() const {

  std::unordered_set<TableId> indexes_to_build;

  {

    auto l = indexed_table_->LockForRead();

    const auto& indexed_table_pb = l.data().pb;

    if (indexed_table_pb.backfill_jobs_size() == 0) {

      // Some other task already marked the backfill job as done.

      return {};

    }

    DCHECK(indexed_table_pb.backfill_jobs_size() == 1) << "For now we only expect to have up to 1 "

                                                          "outstanding backfill job.";

    for (const auto& kv_pair : indexed_table_pb.backfill_jobs(0).backfill_state()) {

      if (kv_pair.second == BackfillJobPB::IN_PROGRESS) {

        indexes_to_build.insert(kv_pair.first);

      }

    }

  }

  return indexes_to_build;

}

void BackfillTable::Launch() {

  backfill_job_ = std::make_shared<BackfillTableJob>(shared_from_this());

  backfill_job_->SetState(MonitoredTaskState::kRunning);

  master_->catalog_manager_impl()->jobs_tracker_->AddTask(backfill_job_);

  {

    auto l = indexed_table_->LockForWrite();

    if (l.data().pb.backfill_jobs_size() == 0) {

      auto* backfill_job = l.mutable_data()->pb.add_backfill_jobs();

      for (const auto& idx_info : index_infos_) {

        backfill_job->add_indexes()->CopyFrom(idx_info);

        backfill_job->mutable_backfill_state()->insert(

            {idx_info.table_id(), BackfillJobPB::IN_PROGRESS});

      }

      auto s = master_->catalog_manager_impl()->sys_catalog_->Upsert(

              leader_term(), indexed_table_);

      if (!s.ok()) {

        LOG(WARNING) << "Failed to persist backfill jobs. Abandoning launch. " << s;

        return;

      }

      l.Commit();

    }

  }

  if (!timestamp_chosen_.load(std::memory_order_acquire)) {

    LaunchComputeSafeTimeForRead();

  } else {

    LaunchBackfill();

  }

}

void BackfillTable::LaunchComputeSafeTimeForRead() {

  auto tablets = indexed_table_->GetTablets();

  num_tablets_.store(tablets.size(), std::memory_order_release);

  tablets_pending_.store(tablets.size(), std::memory_order_release);

  auto min_cutoff = master()->clock()->Now();

  for (const scoped_refptr<TabletInfo>& tablet : tablets) {

    auto get_safetime = std::make_shared<GetSafeTimeForTablet>(

        shared_from_this(), tablet, min_cutoff);

    get_safetime->Launch();

  }

}

std::string BackfillTable::LogPrefix() const {

  return Format("Backfill Index Table(s) $0 ", requested_index_names_);

}

std::string BackfillTable::description() const {

  auto num_pending = tablets_pending_.load(std::memory_order_acquire);

  auto num_tablets = num_tablets_.load(std::memory_order_acquire);

  return Format(

      "Backfill Index Table(s) $0 : $1", requested_index_names_,

      (timestamp_chosen()

           ? (done() ? Format("Backfill $0/$1 tablets done", num_pending, num_tablets)

                     : Format(

                           "Backfilling $0/$1 tablets with $2 rows done", num_pending, num_tablets,

                           number_rows_processed_.load()))

           : Format("Waiting to GetSafeTime from $0/$1 tablets", num_pending, num_tablets)));

}

const std::string BackfillTable::GetNamespaceName() const {

  return ns_info_->name();

}

Status BackfillTable::UpdateRowsProcessedForIndexTable(const uint64_t number_rows_processed) {

  auto l = indexed_table_->LockForWrite();

  if (l.data().pb.backfill_jobs_size() == 0) {

    // Some other task already marked the backfill job as done.

    return Status::OK();

  }

  // This is consistent with logic assuming that we have only one backfill job in queue

  // We might in the future change this to a for loop to account for multiple backfill jobs

  number_rows_processed_.fetch_add(number_rows_processed);

  auto* indexed_table_pb = l.mutable_data()->pb.mutable_backfill_jobs(0);

  indexed_table_pb->set_num_rows_processed(number_rows_processed_.load());

  VLOG(2) << "Updated backfill task to having processed " << number_rows_processed

          << " more rows. Total rows processed is: " << number_rows_processed_;

  RETURN_NOT_OK(

      master_->catalog_manager_impl()->sys_catalog_->Upsert(leader_term(), indexed_table_));

  l.Commit();

  return Status::OK();

}

Status BackfillTable::UpdateSafeTime(const Status& s, HybridTime ht) {

  if (!s.ok()) {

    // Move on to ABORTED permission.

    LOG_WITH_PREFIX(ERROR)

        << "Failed backfill. Could not compute safe time for "

        << yb::ToString(indexed_table_) << " " << s;

    if (!timestamp_chosen_.exchange(true)) {

      RETURN_NOT_OK_PREPEND(

          MarkAllIndexesAsFailed(), "Failed to mark backfill as failed. Abandoning.");

    }

    return Status::OK();

  }

  // Need to guard this.

  HybridTime read_timestamp;

  {

    std::lock_guard<simple_spinlock> l(mutex_);

    VLOG(2) << "Updating read_time_for_backfill_ to max{ "

            << read_time_for_backfill_.ToString() << ", " << ht.ToString()

            << " }.";

    read_time_for_backfill_.MakeAtLeast(ht);

    read_timestamp = read_time_for_backfill_;

  }

  // If OK then move on to doing backfill.

  if (!timestamp_chosen() && 
--tablets_pending_ == 04.39k
) {

    LOG_WITH_PREFIX(INFO) << "Completed fetching SafeTime for the table "

                          << yb::ToString(indexed_table_) << " will be using "

                          << read_timestamp.ToString();

    {

      auto l = indexed_table_->LockForWrite();

      DCHECK_EQ(l.mutable_data()->pb.backfill_jobs_size(), 1);

      auto* backfill_job = l.mutable_data()->pb.mutable_backfill_jobs(0);

      backfill_job->set_backfilling_timestamp(read_timestamp.ToUint64());

      RETURN_NOT_OK_PREPEND(

          master_->catalog_manager_impl()->sys_catalog_->Upsert(

              leader_term(), indexed_table_),

          "Failed to persist backfilling timestamp. Abandoning.");

      l.Commit();

    }

    VLOG_WITH_PREFIX(2) << "Saved " << read_timestamp

                        << " as backfilling_timestamp";

    timestamp_chosen_.store(true, std::memory_order_release);

    LaunchBackfill();

  }

  return Status::OK();

}

void BackfillTable::LaunchBackfill() {

  
VLOG_WITH_PREFIX1
(1) << "launching backfill with timestamp: "

                      << read_time_for_backfill_;

  auto tablets = indexed_table_->GetTablets();

  num_tablets_.store(tablets.size(), std::memory_order_release);

  tablets_pending_.store(tablets.size(), std::memory_order_release);

  for (const scoped_refptr<TabletInfo>& tablet : tablets) {

    auto backfill_tablet = std::make_shared<BackfillTablet>(shared_from_this(), tablet);

    backfill_tablet->Launch();

  }

}

void BackfillTable::Done(const Status& s, const std::unordered_set<TableId>& failed_indexes) {

  if (!s.ok()) {

    LOG_WITH_PREFIX(ERROR) << "failed to backfill the index: " << yb::ToString(failed_indexes)

                           << " due to " << s;

    WARN_NOT_OK(

        MarkIndexesAsFailed(failed_indexes, s.message().ToBuffer()),

        "Couldn't to mark Indexes as failed");

    CheckIfDone();

    return;

  }

  // If OK then move on to READ permissions.

  if (!done() && 
--tablets_pending_ == 04.16k
) {

    LOG_WITH_PREFIX(INFO) << "Completed backfilling the index table.";

    done_.store(true, std::memory_order_release);

    WARN_NOT_OK(MarkAllIndexesAsSuccess(), "Failed to complete backfill.");

    WARN_NOT_OK(UpdateIndexPermissionsForIndexes(), "Failed to complete backfill.");

  } else {

    
VLOG_WITH_PREFIX0
(1) << "Still backfilling " << tablets_pending_ << " more tablets."0
;

  }

}

Status BackfillTable::MarkIndexesAsFailed(

    const std::unordered_set<TableId>& failed_indexes, const string& message) {

  return MarkIndexesAsDesired(failed_indexes, BackfillJobPB::FAILED, message);

}

Status BackfillTable::MarkAllIndexesAsFailed() {

  return MarkIndexesAsDesired(indexes_to_build(), BackfillJobPB::FAILED, "failed");

}

Status BackfillTable::MarkAllIndexesAsSuccess() {

  return MarkIndexesAsDesired(indexes_to_build(), BackfillJobPB::SUCCESS, "");

}

Status BackfillTable::MarkIndexesAsDesired(

    const std::unordered_set<TableId>& index_ids_set, BackfillJobPB_State state,

    const string message) {

  
VLOG_WITH_PREFIX1
(3) << "Marking " << yb::ToString(index_ids_set)

                      << " as " << BackfillJobPB_State_Name(state)

                      << " due to " << message;

  if (!index_ids_set.empty()) {

    auto l = indexed_table_->LockForWrite();

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

    DCHECK_LE(indexed_table_pb.backfill_jobs_size(), 1) << "For now we only expect to have up to 1 "

                                                           "outstanding backfill job.";

    if (indexed_table_pb.backfill_jobs_size() == 0) {

      // Some other task already marked the backfill job as done.

      return Status::OK();

    }

    auto* backfill_state_pb = indexed_table_pb.mutable_backfill_jobs(0)->mutable_backfill_state();

    for (const auto& idx_id : index_ids_set) {

      backfill_state_pb->at(idx_id) = state;

      VLOG
(2) << "Marking index " << idx_id << " as " << BackfillJobPB_State_Name(state)1
;

    }

    for (int i = 0; i < indexed_table_pb.indexes_size(); 
i++1.57k
) {

      IndexInfoPB* idx_pb = indexed_table_pb.mutable_indexes(i);

      if (index_ids_set.find(idx_pb->table_id()) != index_ids_set.end()) {

        // Should this also move to the BackfillJob instead?

        if (!message.empty()) {

          idx_pb->set_backfill_error_message(message);

        } else {

          idx_pb->clear_backfill_error_message();

        }

        idx_pb->clear_is_backfill_deferred();

      }

    }

    RETURN_NOT_OK(

        master_->catalog_manager_impl()->sys_catalog_->Upsert(leader_term(), indexed_table_));

    l.Commit();

  }

  return Status::OK();

}

void BackfillTable::CheckIfDone() {

  if (indexes_to_build().empty()) {

    done_.store(true, std::memory_order_release);

    WARN_NOT_OK(

        UpdateIndexPermissionsForIndexes(), "Could not update index permissions after backfill");

  }

}

Status BackfillTable::UpdateIndexPermissionsForIndexes() {

  std::unordered_map<TableId, IndexPermissions> permissions_to_set;

  bool all_success = true;

  {

    auto l = indexed_table_->LockForRead();

    const auto& indexed_table_pb = l.data().pb;

    if (indexed_table_pb.backfill_jobs_size() == 0) {

      // Some other task already marked the backfill job as done.

      return Status::OK();

    }

    DCHECK(indexed_table_pb.backfill_jobs_size() == 1) << "For now we only expect to have up to 1 "

                                                          "outstanding backfill job.";

    for (const auto& kv_pair : indexed_table_pb.backfill_jobs(0).backfill_state()) {

      VLOG(2) << "Reading backfill_state for " << kv_pair.first << " as "

              << BackfillJobPB_State_Name(kv_pair.second);

      DCHECK_NE(kv_pair.second, BackfillJobPB::IN_PROGRESS)

          << __func__ << " is expected to be only called after all indexes are done.";

      const bool success = (kv_pair.second == BackfillJobPB::SUCCESS);

      all_success &= success;

      permissions_to_set.emplace(

          kv_pair.first,

          success ? 
INDEX_PERM_READ_WRITE_AND_DELETE879
 : 
INDEX_PERM_WRITE_AND_DELETE_WHILE_REMOVING17
);

    }

  }

  for (const auto& kv_pair : permissions_to_set) {

    if (kv_pair.second == INDEX_PERM_READ_WRITE_AND_DELETE) {

      RETURN_NOT_OK(AllowCompactionsToGCDeleteMarkers(kv_pair.first));

    }

  }

  RETURN_NOT_OK_PREPEND(

      MultiStageAlterTable::UpdateIndexPermission(

          master_->catalog_manager_impl(), indexed_table_, permissions_to_set, boost::none),

      "Could not update permissions after backfill. "

      "Possible that the master-leader has changed.");

  backfill_job_->SetState(

      all_success ? 
MonitoredTaskState::kComplete859
 : 
MonitoredTaskState::kFailed25
);

  RETURN_NOT_OK(ClearCheckpointStateInTablets());

  indexed_table_->ClearIsBackfilling();

  VLOG
(1) << "Sending alter table requests to the Indexed table"20
;

  RETURN_NOT_OK(master_->catalog_manager_impl()->SendAlterTableRequest(indexed_table_));

  VLOG
(1) << "DONE Sending alter table requests to the Indexed table"20
;

  LOG(INFO) << "Done backfill on " << indexed_table_->ToString() << " setting permissions to "

            << yb::ToString(permissions_to_set);

  return Status::OK();

}

Status BackfillTable::ClearCheckpointStateInTablets() {

  auto tablets = indexed_table_->GetTablets();

  std::vector<TabletInfo*> tablet_ptrs;

  for (scoped_refptr<TabletInfo>& tablet : tablets) {

    tablet_ptrs.push_back(tablet.get());

    tablet->mutable_metadata()->StartMutation();

    auto& pb = tablet->mutable_metadata()->mutable_dirty()->pb;

    for (const auto& idx : requested_index_ids_) {

      pb.mutable_backfilled_until()->erase(idx);

    }

  }

  RETURN_NOT_OK_PREPEND(

      master()->catalog_manager()->sys_catalog()->Upsert(leader_term(), tablet_ptrs),

      "Could not persist that the table is done backfilling.");

  for (scoped_refptr<TabletInfo>& tablet : tablets) {

    VLOG(2) << "Done backfilling the table. " << yb::ToString(tablet)

            << " clearing backfilled_until";

    tablet->mutable_metadata()->CommitMutation();

  }

  if (FLAGS_TEST_slowdown_backfill_job_deletion_ms > 0) {

    SleepFor(MonoDelta::FromMilliseconds(FLAGS_TEST_slowdown_backfill_job_deletion_ms));

  }

  {

    auto l = indexed_table_->LockForWrite();