/*
 * 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.
 */
package org.apache.ambari.server.upgrade;


import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import jakarta.persistence.EntityManager;
import jakarta.persistence.TypedQuery;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.actionmanager.StageFactory;
import org.apache.ambari.server.agent.ExecutionCommand;
import org.apache.ambari.server.configuration.AmbariServerConfigurationCategory;
import org.apache.ambari.server.configuration.AmbariServerConfigurationKey;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.KerberosHelper;
import org.apache.ambari.server.controller.internal.CalculatedStatus;
import org.apache.ambari.server.orm.DBAccessor;
import org.apache.ambari.server.orm.dao.AmbariConfigurationDAO;
import org.apache.ambari.server.orm.dao.ArtifactDAO;
import org.apache.ambari.server.orm.dao.ClusterServiceDAO;
import org.apache.ambari.server.orm.dao.DaoUtils;
import org.apache.ambari.server.orm.dao.HostComponentDesiredStateDAO;
import org.apache.ambari.server.orm.dao.HostComponentStateDAO;
import org.apache.ambari.server.orm.dao.RequestDAO;
import org.apache.ambari.server.orm.dao.ServiceComponentDesiredStateDAO;
import org.apache.ambari.server.orm.dao.ServiceDesiredStateDAO;
import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
import org.apache.ambari.server.orm.entities.AlertGroupEntity;
import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
import org.apache.ambari.server.orm.entities.ArtifactEntity;
import org.apache.ambari.server.orm.entities.ClusterServiceEntity;
import org.apache.ambari.server.orm.entities.ClusterServiceEntityPK;
import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.HostComponentStateEntity;
import org.apache.ambari.server.orm.entities.RequestEntity;
import org.apache.ambari.server.orm.entities.ServiceComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.ServiceConfigEntity;
import org.apache.ambari.server.orm.entities.ServiceDesiredStateEntity;
import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.security.authorization.UserAuthenticationType;
import org.apache.ambari.server.serveraction.kerberos.KerberosServerAction;
import org.apache.ambari.server.serveraction.kerberos.PrepareKerberosIdentitiesServerAction;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.SecurityType;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.state.kerberos.AbstractKerberosDescriptorContainer;
import org.apache.ambari.server.state.kerberos.KerberosComponentDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosConfigurationDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
import org.apache.ambari.server.state.kerberos.KerberosDescriptorFactory;
import org.apache.ambari.server.state.kerberos.KerberosServiceDescriptor;
import org.apache.ambari.server.topology.validators.HiveServiceValidator;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Sets;
import com.google.common.net.HostAndPort;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonPrimitive;
import com.google.inject.Inject;
import com.google.inject.Injector;

public class UpgradeCatalog270 extends AbstractUpgradeCatalog {

  /**
   * Logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger(UpgradeCatalog270.class);

  protected static final String STAGE_TABLE = "stage";
  protected static final String STAGE_STATUS_COLUMN = "status";
  protected static final String STAGE_DISPLAY_STATUS_COLUMN = "display_status";
  protected static final String REQUEST_TABLE = "request";
  protected static final String REQUEST_DISPLAY_STATUS_COLUMN = "display_status";
  protected static final String REQUEST_USER_NAME_COLUMN = "user_name";
  protected static final String HOST_ROLE_COMMAND_TABLE = "host_role_command";
  protected static final String HRC_OPS_DISPLAY_NAME_COLUMN = "ops_display_name";
  protected static final String COMPONENT_DESIRED_STATE_TABLE = "hostcomponentdesiredstate";
  protected static final String COMPONENT_STATE_TABLE = "hostcomponentstate";
  protected static final String COMPONENT_LAST_STATE_COLUMN = "last_live_state";
  protected static final String SERVICE_DESIRED_STATE_TABLE = "servicedesiredstate";
  protected static final String SECURITY_STATE_COLUMN = "security_state";

  protected static final String AMBARI_SEQUENCES_TABLE = "ambari_sequences";
  protected static final String AMBARI_SEQUENCES_SEQUENCE_NAME_COLUMN = "sequence_name";
  protected static final String AMBARI_SEQUENCES_SEQUENCE_VALUE_COLUMN = "sequence_value";

  protected static final String AMBARI_CONFIGURATION_TABLE = "ambari_configuration";
  protected static final String AMBARI_CONFIGURATION_CATEGORY_NAME_COLUMN = "category_name";
  protected static final String AMBARI_CONFIGURATION_PROPERTY_NAME_COLUMN = "property_name";
  protected static final String AMBARI_CONFIGURATION_PROPERTY_VALUE_COLUMN = "property_value";

  protected static final String USER_AUTHENTICATION_TABLE = "user_authentication";
  protected static final String USER_AUTHENTICATION_USER_AUTHENTICATION_ID_COLUMN = "user_authentication_id";
  protected static final String USER_AUTHENTICATION_USER_ID_COLUMN = "user_id";
  protected static final String USER_AUTHENTICATION_AUTHENTICATION_TYPE_COLUMN = "authentication_type";
  protected static final String USER_AUTHENTICATION_AUTHENTICATION_KEY_COLUMN = "authentication_key";
  protected static final String USER_AUTHENTICATION_CREATE_TIME_COLUMN = "create_time";
  protected static final String USER_AUTHENTICATION_UPDATE_TIME_COLUMN = "update_time";
  protected static final String USER_AUTHENTICATION_PRIMARY_KEY = "PK_user_authentication";
  protected static final String USER_AUTHENTICATION_USER_AUTHENTICATION_USER_ID_INDEX = "IDX_user_authentication_user_id";
  protected static final String USER_AUTHENTICATION_USER_AUTHENTICATION_USERS_FOREIGN_KEY = "FK_user_authentication_users";

  protected static final String USERS_TABLE = "users";
  protected static final String USERS_USER_ID_COLUMN = "user_id";
  protected static final String USERS_PRINCIPAL_ID_COLUMN = "principal_id";
  protected static final String USERS_USER_TYPE_COLUMN = "user_type";
  protected static final String USERS_USER_PASSWORD_COLUMN = "user_password";
  protected static final String USERS_CREATE_TIME_COLUMN = "create_time";
  protected static final String USERS_LDAP_USER_COLUMN = "ldap_user";
  protected static final String USERS_CONSECUTIVE_FAILURES_COLUMN = "consecutive_failures";
  protected static final String USERS_USER_NAME_COLUMN = "user_name";
  protected static final String USERS_DISPLAY_NAME_COLUMN = "display_name";
  protected static final String USERS_LOCAL_USERNAME_COLUMN = "local_username";
  protected static final String USERS_VERSION_COLUMN = "version";
  protected static final String UNIQUE_USERS_0_INDEX = "UNQ_users_0";

  protected static final String MEMBERS_TABLE = "members";
  protected static final String MEMBERS_MEMBER_ID_COLUMN = "member_id";
  protected static final String MEMBERS_GROUP_ID_COLUMN = "group_id";
  protected static final String MEMBERS_USER_ID_COLUMN = "user_id";

  protected static final String ADMINPRIVILEGE_TABLE = "adminprivilege";
  protected static final String ADMINPRIVILEGE_PRIVILEGE_ID_COLUMN = "privilege_id";
  protected static final String ADMINPRIVILEGE_PERMISSION_ID_COLUMN = "permission_id";
  protected static final String ADMINPRIVILEGE_RESOURCE_ID_COLUMN = "resource_id";
  protected static final String ADMINPRIVILEGE_PRINCIPAL_ID_COLUMN = "principal_id";

  // kerberos tables constants
  protected static final String KERBEROS_KEYTAB_TABLE = "kerberos_keytab";
  protected static final String KERBEROS_KEYTAB_PRINCIPAL_TABLE = "kerberos_keytab_principal";
  protected static final String KKP_MAPPING_SERVICE_TABLE = "kkp_mapping_service";
  protected static final String KEYTAB_PATH_FIELD = "keytab_path";
  protected static final String OWNER_NAME_FIELD = "owner_name";
  protected static final String OWNER_ACCESS_FIELD = "owner_access";
  protected static final String GROUP_NAME_FIELD = "group_name";
  protected static final String GROUP_ACCESS_FIELD = "group_access";
  protected static final String IS_AMBARI_KEYTAB_FIELD = "is_ambari_keytab";
  protected static final String WRITE_AMBARI_JAAS_FIELD = "write_ambari_jaas";
  protected static final String PK_KERBEROS_KEYTAB = "PK_kerberos_keytab";
  protected static final String KKP_ID_COLUMN = "kkp_id";
  protected static final String PRINCIPAL_NAME_COLUMN = "principal_name";
  protected static final String IS_DISTRIBUTED_COLUMN = "is_distributed";
  protected static final String PK_KKP = "PK_kkp";
  protected static final String UNI_KKP = "UNI_kkp";
  protected static final String SERVICE_NAME_COLUMN = "service_name";
  protected static final String COMPONENT_NAME_COLUMN = "component_name";
  protected static final String PK_KKP_MAPPING_SERVICE = "PK_kkp_mapping_service";
  protected static final String FK_KKP_KEYTAB_PATH = "FK_kkp_keytab_path";
  protected static final String FK_KKP_HOST_ID = "FK_kkp_host_id";
  protected static final String FK_KKP_PRINCIPAL_NAME = "FK_kkp_principal_name";
  protected static final String HOSTS_TABLE = "hosts";
  protected static final String KERBEROS_PRINCIPAL_TABLE = "kerberos_principal";
  protected static final String FK_KKP_SERVICE_PRINCIPAL = "FK_kkp_service_principal";
  protected static final String KKP_ID_SEQ_NAME = "kkp_id_seq";
  protected static final String KERBEROS_PRINCIPAL_HOST_TABLE = "kerberos_principal_host";
  protected static final String HOST_ID_COLUMN = "host_id";

  protected static final String REPO_OS_TABLE = "repo_os";
  protected static final String REPO_OS_ID_COLUMN = "id";
  protected static final String REPO_OS_REPO_VERSION_ID_COLUMN = "repo_version_id";
  protected static final String REPO_OS_FAMILY_COLUMN = "family";
  protected static final String REPO_OS_AMBARI_MANAGED_COLUMN = "ambari_managed";
  protected static final String REPO_OS_PRIMARY_KEY = "PK_repo_os_id";
  protected static final String REPO_OS_FOREIGN_KEY = "FK_repo_os_id_repo_version_id";

  protected static final String REPO_DEFINITION_TABLE = "repo_definition";
  protected static final String REPO_DEFINITION_ID_COLUMN = "id";
  protected static final String REPO_DEFINITION_REPO_OS_ID_COLUMN = "repo_os_id";
  protected static final String REPO_DEFINITION_REPO_NAME_COLUMN = "repo_name";
  protected static final String REPO_DEFINITION_REPO_ID_COLUMN = "repo_id";
  protected static final String REPO_DEFINITION_BASE_URL_COLUMN = "base_url";
  protected static final String REPO_DEFINITION_DISTRIBUTION_COLUMN = "distribution";
  protected static final String REPO_DEFINITION_COMPONENTS_COLUMN = "components";
  protected static final String REPO_DEFINITION_UNIQUE_REPO_COLUMN = "unique_repo";
  protected static final String REPO_DEFINITION_MIRRORS_COLUMN = "mirrors";
  protected static final String REPO_DEFINITION_PRIMARY_KEY = "PK_repo_definition_id";
  protected static final String REPO_DEFINITION_FOREIGN_KEY = "FK_repo_definition_repo_os_id";

  protected static final String REPO_TAGS_TABLE = "repo_tags";
  protected static final String REPO_TAGS_REPO_DEFINITION_ID_COLUMN = "repo_definition_id";
  protected static final String REPO_TAGS_TAG_COLUMN = "tag";
  protected static final String REPO_TAGS_FOREIGN_KEY = "FK_repo_tag_definition_id";

  protected static final String REPO_APPLICABLE_SERVICES_TABLE = "repo_applicable_services";
  protected static final String REPO_APPLICABLE_SERVICES_REPO_DEFINITION_ID_COLUMN = "repo_definition_id";
  protected static final String REPO_APPLICABLE_SERVICES_SERVICE_NAME_COLUMN = "service_name";
  protected static final String REPO_APPLICABLE_SERVICES_FOREIGN_KEY = "FK_repo_app_service_def_id";

  protected static final String REPO_VERSION_TABLE = "repo_version";
  protected static final String REPO_VERSION_REPO_VERSION_ID_COLUMN = "repo_version_id";
  protected static final String REPO_VERSION_REPOSITORIES_COLUMN = "repositories";

  protected static final String WIDGET_TABLE = "widget";
  protected static final String WIDGET_TAG_COLUMN = "tag";

  protected static final String SERVICE_COMPONENT_DESIRED_STATE_TABLE = "servicecomponentdesiredstate";
  protected static final String HIVE_SERVICE_COMPONENT_WEBHCAT_SERVER = "WEBHCAT_SERVER";
  protected static final String CONFIGURATION_CORE_SITE = "core-site";
  protected static final String CONFIGURATION_WEBHCAT_SITE = "webhcat-site";
  protected static final String PROPERTY_HADOOP_PROXYUSER_HTTP_HOSTS = "hadoop.proxyuser.HTTP.hosts";
  protected static final String PROPERTY_TEMPLETON_HIVE_PROPERTIES = "templeton.hive.properties";
  public static final String AMBARI_INFRA_OLD_NAME = "AMBARI_INFRA";
  public static final String AMBARI_INFRA_NEW_NAME = "AMBARI_INFRA_SOLR";

  public static final String SERVICE_CONFIG_MAPPING_TABLE = "serviceconfigmapping";
  public static final String CLUSTER_CONFIG_TABLE = "clusterconfig";

  // Broken constraints added by Views
  public static final String FK_HOSTCOMPONENTDESIREDSTATE_COMPONENT_NAME = "fk_hostcomponentdesiredstate_component_name";
  public static final String FK_HOSTCOMPONENTSTATE_COMPONENT_NAME = "fk_hostcomponentstate_component_name";
  public static final String FK_SERVICECOMPONENTDESIREDSTATE_SERVICE_NAME = "fk_servicecomponentdesiredstate_service_name";

  static final String YARN_SERVICE = "YARN";

  @Inject
  DaoUtils daoUtils;

  // ----- Constructors ------------------------------------------------------

  /**
   * Don't forget to register new UpgradeCatalogs in {@link org.apache.ambari.server.upgrade.SchemaUpgradeHelper.UpgradeHelperModule#configure()}
   *
   * @param injector Guice injector to track dependencies and uses bindings to inject them.
   */
  @Inject
  public UpgradeCatalog270(Injector injector) {
    super(injector);

    daoUtils = injector.getInstance(DaoUtils.class);
  }

  // ----- UpgradeCatalog ----------------------------------------------------

  /**
   * {@inheritDoc}
   */
  @Override
  public String getTargetVersion() {
    return "2.7.0";
  }

  // ----- AbstractUpgradeCatalog --------------------------------------------

  /**
   * {@inheritDoc}
   */
  @Override
  public String getSourceVersion() {
    return "2.6.2";
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void executeDDLUpdates() throws AmbariException, SQLException {
    dropBrokenFKs();
    updateStageTable();
    updateRequestTable();
    addOpsDisplayNameColumnToHostRoleCommand();
    removeSecurityState();
    addAmbariConfigurationTable();
    addHostComponentLastStateTable();
    upgradeUserTables();
    upgradeKerberosTables();
    upgradeRepoTables();
    upgradeWidgetTable();
  }

  /**
   * Upgrade the users table as well as supporting tables.
   * <p>
   * Affected table are
   * <ul>
   * <li>users</li>
   * <li>user_authentication (new)</li>
   * <li>members</li>
   * <li>adminprivilege</li>
   * </ul>
   *
   * @throws SQLException if an error occurs while executing SQL statements
   * @see #createUserAuthenticationTable()
   * @see #updateGroupMembershipRecords()
   * @see #updateAdminPrivilegeRecords()
   * @see #updateUsersTable()
   */
  protected void upgradeUserTables() throws SQLException {
    convertUserCreationTimeToLong();
    createUserAuthenticationTable();
    updateGroupMembershipRecords();
    updateAdminPrivilegeRecords();
    updateUsersTable();
  }

  protected void upgradeRepoTables() throws SQLException {
    createRepoOsTable();
    createRepoDefinitionTable();
    createRepoTagsTable();
    createRepoApplicableServicesTable();
    migrateRepoData();
    updateRepoVersionTable();
  }

  /**
   * Adds the repo_os table to the Ambari database.
   * <pre>
   * CREATE TABLE repo_os (
   *   id BIGINT NOT NULL,
   *   repo_version_id BIGINT NOT NULL,
   *   family VARCHAR(255) NOT NULL DEFAULT '',
   *   ambari_managed SMALLINT DEFAULT 1,
   *   CONSTRAINT PK_repo_os_id PRIMARY KEY (id),
   *   CONSTRAINT FK_repo_os_id_repo_version_id FOREIGN KEY (repo_version_id) REFERENCES repo_version (repo_version_id));
   * </pre>
   *
   * @throws SQLException
   */
  private void createRepoOsTable() throws SQLException {
    List<DBAccessor.DBColumnInfo> columns = new ArrayList<>();
    columns.add(new DBAccessor.DBColumnInfo(REPO_OS_ID_COLUMN, Long.class, null, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_OS_REPO_VERSION_ID_COLUMN, Long.class, null, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_OS_FAMILY_COLUMN, String.class, 255, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_OS_AMBARI_MANAGED_COLUMN, Integer.class, null, 1, true));

    dbAccessor.createTable(REPO_OS_TABLE, columns);
    dbAccessor.addPKConstraint(REPO_OS_TABLE, REPO_OS_PRIMARY_KEY, REPO_OS_ID_COLUMN);
    dbAccessor.addFKConstraint(REPO_OS_TABLE, REPO_OS_FOREIGN_KEY, REPO_OS_REPO_VERSION_ID_COLUMN, REPO_VERSION_TABLE, REPO_VERSION_REPO_VERSION_ID_COLUMN, false);
  }

  /**
   * Adds the repo_definition table to the Ambari database.
   * <pre>
   *   CREATE TABLE repo_definition (
   *     id BIGINT NOT NULL,
   *     repo_os_id BIGINT,
   *     repo_name VARCHAR(255) NOT NULL,
   *     repo_id VARCHAR(255) NOT NULL,
   *     base_url VARCHAR(2048) NOT NULL,
   *     distribution VARCHAR(2048),
   *     components VARCHAR(2048),
   *     unique_repo SMALLINT DEFAULT 1,
   *     mirrors VARCHAR(2048),
   *     CONSTRAINT PK_repo_definition_id PRIMARY KEY (id),
   *     CONSTRAINT FK_repo_definition_repo_os_id FOREIGN KEY (repo_os_id) REFERENCES repo_os (id));
   * </pre>
   *
   * @throws SQLException
   */
  private void createRepoDefinitionTable() throws SQLException {
    List<DBAccessor.DBColumnInfo> columns = new ArrayList<>();
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_ID_COLUMN, Long.class, null, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_REPO_OS_ID_COLUMN, Long.class, null, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_REPO_NAME_COLUMN, String.class, 255, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_REPO_ID_COLUMN, String.class, 255, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_BASE_URL_COLUMN, String.class, 2048, null, true));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_DISTRIBUTION_COLUMN, String.class, 2048, null, true));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_COMPONENTS_COLUMN, String.class, 2048, null, true));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_UNIQUE_REPO_COLUMN, Integer.class, 1, 1, true));
    columns.add(new DBAccessor.DBColumnInfo(REPO_DEFINITION_MIRRORS_COLUMN, String.class, 2048, null, true));

    dbAccessor.createTable(REPO_DEFINITION_TABLE, columns);
    dbAccessor.addPKConstraint(REPO_DEFINITION_TABLE, REPO_DEFINITION_PRIMARY_KEY, REPO_DEFINITION_ID_COLUMN);
    dbAccessor.addFKConstraint(REPO_DEFINITION_TABLE, REPO_DEFINITION_FOREIGN_KEY, REPO_DEFINITION_REPO_OS_ID_COLUMN, REPO_OS_TABLE, REPO_OS_ID_COLUMN, false);
  }

  /**
   * Adds the repo_tags table to the Ambari database.
   * <pre>
   *   CREATE TABLE repo_tags (
   *     repo_definition_id BIGINT NOT NULL,
   *     tag VARCHAR(255) NOT NULL,
   *     CONSTRAINT FK_repo_tag_definition_id FOREIGN KEY (repo_definition_id) REFERENCES repo_definition (id));
   * </pre>
   *
   * @throws SQLException
   */
  private void createRepoTagsTable() throws SQLException {
    List<DBAccessor.DBColumnInfo> columns = new ArrayList<>();
    columns.add(new DBAccessor.DBColumnInfo(REPO_TAGS_REPO_DEFINITION_ID_COLUMN, Long.class, null, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_TAGS_TAG_COLUMN, String.class, 255, null, false));

    dbAccessor.createTable(REPO_TAGS_TABLE, columns);
    dbAccessor.addFKConstraint(REPO_TAGS_TABLE, REPO_TAGS_FOREIGN_KEY, REPO_TAGS_REPO_DEFINITION_ID_COLUMN, REPO_DEFINITION_TABLE, REPO_DEFINITION_ID_COLUMN, false);
  }

  /**
   * Adds the repo_applicable_services table to the Ambari database.
   * <pre>
   *   CREATE TABLE repo_applicable_services (
   *     repo_definition_id BIGINT NOT NULL,
   *     service_name VARCHAR(255) NOT NULL,
   *     CONSTRAINT FK_repo_applicable_service_definition_id FOREIGN KEY (repo_definition_id) REFERENCES repo_definition (id));
   * </pre>
   *
   * @throws SQLException
   */
  private void createRepoApplicableServicesTable() throws SQLException {
    List<DBAccessor.DBColumnInfo> columns = new ArrayList<>();
    columns.add(new DBAccessor.DBColumnInfo(REPO_APPLICABLE_SERVICES_REPO_DEFINITION_ID_COLUMN, Long.class, null, null, false));
    columns.add(new DBAccessor.DBColumnInfo(REPO_APPLICABLE_SERVICES_SERVICE_NAME_COLUMN, String.class, 255, null, false));

    dbAccessor.createTable(REPO_APPLICABLE_SERVICES_TABLE, columns);
    dbAccessor.addFKConstraint(REPO_APPLICABLE_SERVICES_TABLE, REPO_APPLICABLE_SERVICES_FOREIGN_KEY, REPO_APPLICABLE_SERVICES_REPO_DEFINITION_ID_COLUMN, REPO_DEFINITION_TABLE, REPO_DEFINITION_ID_COLUMN, false);
  }

  /**
   * Perform steps to move data from the old repo_version.repositories structure into new tables -
   * repo_os, repo_definition, repo_tags
   *
   * @throws SQLException
   */
  private void migrateRepoData() throws SQLException {
    if(dbAccessor.tableHasColumn(REPO_VERSION_TABLE, REPO_VERSION_REPOSITORIES_COLUMN)) {
      int repoOsId = 0;
      int repoDefinitionId = 0;

      // Get a map of repo_version.id to repo_version.repositories
      Map<Long, String> repoVersionData = dbAccessor.getKeyToStringColumnMap(REPO_VERSION_TABLE,
          REPO_VERSION_REPO_VERSION_ID_COLUMN, REPO_VERSION_REPOSITORIES_COLUMN, null, null, true);

      if (repoVersionData != null) {
        // For each entry in the map, parse the repo_version.repositories data and created records in the new
        // repo_os, repo_definition, and repo_tabs tables...
        for (Map.Entry<Long, String> entry : repoVersionData.entrySet()) {
          Long repoVersionId = entry.getKey();
          String repositoriesJson = entry.getValue();

          if (!StringUtils.isEmpty(repositoriesJson)) {
            JsonArray rootJson = new JsonParser().parse(repositoriesJson).getAsJsonArray();

            if (rootJson != null) {
              for (JsonElement rootElement : rootJson) {
                // process each OS element
                JsonObject rootObject = rootElement.getAsJsonObject();

                if (rootObject != null) {
                  JsonPrimitive osType = rootObject.getAsJsonPrimitive("OperatingSystems/os_type");
                  JsonPrimitive ambariManaged = rootObject.getAsJsonPrimitive("OperatingSystems/ambari_managed_repositories");
                  String isAmbariManaged = ambariManaged == null ? "1" :  (ambariManaged.getAsBoolean() ? "1" : "0"); //the SQL script which creates the DB schema defaults to 1

                  JsonArray repositories = rootObject.getAsJsonArray("repositories");

                  dbAccessor.insertRowIfMissing(REPO_OS_TABLE,
                      new String[]{REPO_OS_ID_COLUMN, REPO_OS_REPO_VERSION_ID_COLUMN, REPO_OS_AMBARI_MANAGED_COLUMN, REPO_OS_FAMILY_COLUMN},
                      new String[]{String.valueOf(++repoOsId), String.valueOf(repoVersionId), isAmbariManaged, getFormattedJSONPrimitiveString(osType)},
                      false);

                  if (repositories != null) {
                    for (JsonElement repositoryElement : repositories) {
                      JsonObject repositoryObject = repositoryElement.getAsJsonObject();

                      if (repositoryObject != null) {
                        JsonPrimitive repoId = repositoryObject.getAsJsonPrimitive("Repositories/repo_id");
                        JsonPrimitive repoName = repositoryObject.getAsJsonPrimitive("Repositories/repo_name");
                        JsonPrimitive baseUrl = repositoryObject.getAsJsonPrimitive("Repositories/base_url");
                        JsonArray tags = repositoryObject.getAsJsonArray("Repositories/tags");

                        dbAccessor.insertRowIfMissing(REPO_DEFINITION_TABLE,
                            new String[]{REPO_DEFINITION_ID_COLUMN, REPO_DEFINITION_REPO_OS_ID_COLUMN,
                                REPO_DEFINITION_REPO_NAME_COLUMN, REPO_DEFINITION_REPO_ID_COLUMN, REPO_DEFINITION_BASE_URL_COLUMN},
                            new String[]{String.valueOf(++repoDefinitionId), String.valueOf(repoOsId),
                                getFormattedJSONPrimitiveString(repoName), getFormattedJSONPrimitiveString(repoId),
                                getFormattedJSONPrimitiveString(baseUrl)},
                            false);

                        if (tags != null) {
                          for (JsonElement tagsElement : tags) {
                            JsonPrimitive tag = tagsElement.getAsJsonPrimitive();

                            if (tag != null) {
                              dbAccessor.insertRowIfMissing(REPO_TAGS_TABLE,
                                  new String[]{REPO_TAGS_REPO_DEFINITION_ID_COLUMN, REPO_TAGS_TAG_COLUMN},
                                  new String[]{String.valueOf(repoDefinitionId), getFormattedJSONPrimitiveString(tag)},
                                  false);
                            }
                          }
                        }
                      }
                    }
                  }
                }
              }
            }
          }
        }
      }

      // Add the relevant records in the ambari_sequence table
      // - repo_os_id_seq
      // - repo_definition_id_seq
      dbAccessor.insertRowIfMissing(AMBARI_SEQUENCES_TABLE,
          new String[]{AMBARI_SEQUENCES_SEQUENCE_NAME_COLUMN, AMBARI_SEQUENCES_SEQUENCE_VALUE_COLUMN},
          new String[]{"'repo_os_id_seq'", String.valueOf(++repoOsId)},
          false);
      dbAccessor.insertRowIfMissing(AMBARI_SEQUENCES_TABLE,
          new String[]{AMBARI_SEQUENCES_SEQUENCE_NAME_COLUMN, AMBARI_SEQUENCES_SEQUENCE_VALUE_COLUMN},
          new String[]{"'repo_definition_id_seq'", String.valueOf(++repoDefinitionId)},
          false);
    }
  }

  private String getFormattedJSONPrimitiveString(JsonPrimitive jsonValue) {
    return jsonValue == null ? null : String.format("'%s'", jsonValue.getAsString());
  }

  /**
   * Updates the repo_version table by removing old columns
   *
   * @throws SQLException
   */
  private void updateRepoVersionTable() throws SQLException {
    dbAccessor.dropColumn(REPO_VERSION_TABLE, REPO_VERSION_REPOSITORIES_COLUMN);
  }

  /**
   * In order to save the epoch equivalent of users.create_time we need to convert data in this column as follows:
   * <ol>
   * <li>creating a temporary column where we store the numeric representation of
   * the timestamp
   * <li>populating data in the temporary column
   * <li>removing original column column
   * <li>renaming the temporary column to the original column
   * </ol>
   *
   * @throws SQLException
   *           if an error occurs while executing SQL statements
   *
   */
  private void convertUserCreationTimeToLong() throws SQLException {
    if (!isUserCreationTimeMigrated()) {
      LOG.info("Converting user creation times...");
      final String temporaryColumnName = USERS_CREATE_TIME_COLUMN + "_numeric";
      if (!dbAccessor.tableHasColumn(USERS_TABLE, temporaryColumnName)) {
        final DBAccessor.DBColumnInfo tempColumnInfo = new DBAccessor.DBColumnInfo(temporaryColumnName, Long.class);
        dbAccessor.addColumn(USERS_TABLE, tempColumnInfo);
      }

      if (dbAccessor.tableHasColumn(USERS_TABLE, USERS_CREATE_TIME_COLUMN)) {
        final Map<Integer, Timestamp> currentUserCreateTimes = fetchCurrentUserCreateTimesNotYetMigrated(temporaryColumnName);
        for (Map.Entry<Integer, Timestamp> currentUserCreateTime : currentUserCreateTimes.entrySet()) {
          dbAccessor.updateTable(USERS_TABLE, temporaryColumnName, currentUserCreateTime.getValue().getTime(),
              "WHERE " + USERS_USER_ID_COLUMN + "=" + currentUserCreateTime.getKey());
        }

        dbAccessor.dropColumn(USERS_TABLE, USERS_CREATE_TIME_COLUMN);
      }

      final DBAccessor.DBColumnInfo usersCreateTimeColumnInfo = new DBAccessor.DBColumnInfo(USERS_CREATE_TIME_COLUMN, Long.class, null, null, false);
      dbAccessor.renameColumn(USERS_TABLE, temporaryColumnName, usersCreateTimeColumnInfo);
      LOG.info("Converted user creation times");
    } else {
      LOG.info("Already converted user creation timestamps to EPOCH representation");
    }
  }

  private boolean isUserCreationTimeMigrated() throws SQLException {
    final int columnType = dbAccessor.getColumnType(USERS_TABLE, USERS_CREATE_TIME_COLUMN);
    LOG.info(USERS_TABLE + "." + USERS_CREATE_TIME_COLUMN + "'s type = " + columnType);
    return columnType != Types.DATE && columnType != Types.TIMESTAMP;
  }

  private Map<Integer, Timestamp> fetchCurrentUserCreateTimesNotYetMigrated(String temporaryColumnName) throws SQLException {
    final Map<Integer, Timestamp> currentUserCreateTimes = new HashMap<>();
    try (
        PreparedStatement pstmt = dbAccessor.getConnection().prepareStatement("SELECT " + USERS_USER_ID_COLUMN + ", " + USERS_CREATE_TIME_COLUMN + " FROM " + USERS_TABLE + " WHERE " + temporaryColumnName + " IS NULL ORDER BY " + USERS_USER_ID_COLUMN);
        ResultSet rs = pstmt.executeQuery()) {
      while (rs.next()) {
        currentUserCreateTimes.put(rs.getInt(1), rs.getTimestamp(2) == null ? new Timestamp(System.currentTimeMillis()) : rs.getTimestamp(2));
      }
    }
    return currentUserCreateTimes;
  }

  /**
   * If the <code>users</code> table has not yet been migrated, create the <code>user_authentication</code>
   * table and generate relevant records for that table based on data in the <code>users</code> table.
   * <p>
   * The records in the new <code>user_authentication</code> table represent all of the types associated
   * with a given (case-insensitive) username.  If <code>UserA:LOCAL</code>, <code>usera:LOCAL</code> and
   * <code>usera:LDAP</code> exist in the original <code>users</code> table, three records will be created
   * in the <code>user_authentication</code> table: one for each t
   * to <code>Role1</code>, the three <code>adminprivilege</code> records will be merged into a single
   * record for <code>usera</code>.
   *
   * @throws SQLException if an error occurs while executing SQL statements
   */
