absl::StatusOr SchemaUpdaterImpl::CreateIndexHelper()

in backend/schema/updater/schema_updater.cc [3806:3911]
90 lines of code
17 McCabe index (conditional complexity)

absl::StatusOr<const Index*> SchemaUpdaterImpl::CreateIndexHelper(
    const std::string& index_name, const std::string& index_base_name,
    bool is_unique, bool is_null_filtered,
    const std::string* interleave_in_table,
    const std::vector<ddl::KeyPartClause>& table_pk,
    const ::google::protobuf::RepeatedPtrField<ddl::StoredColumnDefinition>&
        stored_columns,
    bool is_search_index, bool is_vector_index,
    const ::google::protobuf::RepeatedPtrField<ddl::KeyPartClause>* partition_by,
    const ::google::protobuf::RepeatedPtrField<ddl::KeyPartClause>* order_by,
    const ::google::protobuf::RepeatedPtrField<std::string>* null_filtered_columns,
    const ::google::protobuf::RepeatedPtrField<ddl::SetOption>* set_options,
    const Table* indexed_table) {
  if (indexed_table == nullptr) {
    indexed_table = latest_schema_->FindTableCaseSensitive(index_base_name);
    if (indexed_table == nullptr) {
      return error::TableNotFound(index_base_name);
    }
  }
  if (latest_schema_->num_index() >= limits::kMaxIndexesPerDatabase) {
    return error::TooManyIndicesPerDatabase(index_name,
                                            limits::kMaxIndexesPerDatabase);
  }

  // Tables and indexes share a namespace.
  ZETASQL_RETURN_IF_ERROR(global_names_.AddName("Index", index_name));

  Index::Builder builder;
  std::optional<uint32_t> oid = pg_oid_assigner_->GetNextPostgresqlOid();
  builder.set_postgresql_oid(oid);
  if (oid.has_value()) {
    ZETASQL_VLOG(2) << "Assigned oid " << oid.value() << " to index " << index_name;
  }
  builder.set_name(index_name);
  builder.set_unique(is_unique);
  builder.set_null_filtered(is_null_filtered);
  ColumnsUsedByIndex columns_used_by_index;
  ZETASQL_ASSIGN_OR_RETURN(
      std::unique_ptr<const Table> data_table,
      CreateIndexDataTable(index_name, table_pk, interleave_in_table,
                           stored_columns, partition_by, order_by,
                           null_filtered_columns, builder.get(), indexed_table,
                           &columns_used_by_index));

  builder.set_index_data_table(data_table.get());

  for (const KeyColumn* key_col : columns_used_by_index.index_key_columns) {
    builder.add_key_column(key_col);
  }

  for (const Column* col : columns_used_by_index.stored_columns) {
    builder.add_stored_column(col);
  }
  for (const Column* col : columns_used_by_index.null_filtered_columns) {
    builder.add_null_filtered_column(col);
  }
  if (is_search_index) {
    builder.set_index_type(is_search_index);

    for (const Column* col : columns_used_by_index.partition_by_columns) {
      builder.add_partition_by_column(col);
    }

    for (const Column* col : columns_used_by_index.order_by_columns) {
      builder.add_order_by_column(col);
    }
  }
  if (is_vector_index) {
    builder.set_vector_index_type(is_vector_index);
    ZETASQL_RETURN_IF_ERROR(SetVectorIndexOptions(index_name, *set_options, &builder));
  }

  ZETASQL_RETURN_IF_ERROR(AlterNode<Table>(
      indexed_table, [&builder](Table::Editor* table_editor) -> absl::Status {
        table_editor->add_index(builder.get());
        builder.set_indexed_table(table_editor->get());
        return absl::OkStatus();
      }));
  // Register a backfill action for the index.
  const Index* index = builder.get();

  if (!is_search_index && !is_vector_index) {
    statement_context_->AddAction(
        [index](const SchemaValidationContext* context) {
          return BackfillIndex(index, context);
        });
  }

  if (SDLObjectName::IsFullyQualifiedName(index_name)) {
    ZETASQL_RETURN_IF_ERROR(AlterInNamedSchema(
        index_name, [&builder](NamedSchema::Editor* editor) -> absl::Status {
          editor->add_index(builder.get());
          return absl::OkStatus();
        }));
  }

  if (set_options != nullptr && !set_options->empty()) {
    ZETASQL_RETURN_IF_ERROR(SetIndexOptions(*set_options, &builder, is_vector_index));
  }

  // The data table must be added after the index for correct order of
  // validation.
  ZETASQL_RETURN_IF_ERROR(AddNode(builder.build()));
  ZETASQL_RETURN_IF_ERROR(AddNode(std::move(data_table)));
  return index;
}