AMBARI-4874. Fixes for java upgrade feature. (mpapirkovskyy)

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java

@@ -64,6 +64,7 @@ import org.apache.ambari.server.security.unsecured.rest.CertificateDownload;
 import org.apache.ambari.server.security.unsecured.rest.CertificateSign;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.utils.StageUtils;
+import org.apache.ambari.server.utils.VersionUtils;
 import org.apache.ambari.server.view.ViewInstanceDefinition;
 import org.apache.ambari.server.view.ViewRegistry;
 import org.eclipse.jetty.server.Connector;
@@ -464,7 +465,7 @@ public class AmbariServer {
       serverVersion = ambariMetaInfo.getServerVersion();
     }
 
-    if (schemaVersionEntity==null || !schemaVersion.equals(serverVersion)) {
+    if (schemaVersionEntity==null || VersionUtils.compareVersions(schemaVersion, serverVersion, 3) != 0) {
       String error = "Current database store version is not compatible with " +
           "current server version"
           + ", serverVersion=" + serverVersion

ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessor.java

@@ -21,6 +21,7 @@ import org.eclipse.persistence.sessions.DatabaseSession;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
 
@@ -153,6 +154,14 @@ public interface DBAccessor {
    */
   public void executeQuery(String query) throws SQLException;
 
+  /**
+   * Execute select query
+   * @param query
+   * @return
+   * @throws SQLException
+   */
+  ResultSet executeSelect(String query) throws SQLException;
+
   /**
    * Execute query on DB
    * @param query

ambari-server/src/main/java/org/apache/ambari/server/orm/DBAccessorImpl.java

@@ -110,7 +110,7 @@ public class DBAccessorImpl implements DBAccessor {
     }
   }
 
-  private DatabaseMetaData getDatabaseMetaData() throws SQLException {
+  protected DatabaseMetaData getDatabaseMetaData() throws SQLException {
     if (databaseMetaData == null) {
       databaseMetaData = connection.getMetaData();
     }
@@ -118,6 +118,17 @@ public class DBAccessorImpl implements DBAccessor {
     return databaseMetaData;
   }
 
+  private String convertObjectName(String objectName) throws SQLException {
+    DatabaseMetaData metaData = getDatabaseMetaData();
+    if (metaData.storesLowerCaseIdentifiers()) {
+      return objectName.toLowerCase();
+    }else if (metaData.storesUpperCaseIdentifiers()) {
+      return objectName.toUpperCase();
+    }
+
+    return objectName;
+  }
+
   @Override
   public boolean tableExists(String tableName) throws SQLException {
     boolean result = false;
@@ -129,7 +140,7 @@ public class DBAccessorImpl implements DBAccessor {
     }
 
     ResultSet res = metaData.getTables(null, schemaFilter,
-      tableName.toUpperCase(), new String[] { "TABLE" });
+      convertObjectName(tableName), new String[] { "TABLE" });
 
     if (res != null) {
       try {
@@ -172,7 +183,7 @@ public class DBAccessorImpl implements DBAccessor {
     boolean retVal = false;
     if (rs != null) {
       if (rs.next()) {
-        return rs.getInt(0) > 0;
+        return rs.getInt(1) > 0;
       }
     }
     return retVal;
@@ -189,7 +200,7 @@ public class DBAccessorImpl implements DBAccessor {
     }
 
     ResultSet rs = metaData.getColumns(null, schemaFilter,
-        tableName.toUpperCase(), columnName.toUpperCase());
+        convertObjectName(tableName), convertObjectName(columnName));
 
     if (rs != null) {
       try {
@@ -216,14 +227,16 @@ public class DBAccessorImpl implements DBAccessor {
       schemaFilter = configuration.getDatabaseUser();
     }
 
-    ResultSet rs = metaData.getCrossReference(null, schemaFilter, tableName,
-      null, schemaFilter, refTableName);
+    ResultSet rs = metaData.getCrossReference(null, schemaFilter, convertObjectName(tableName),
+      null, schemaFilter, convertObjectName(refTableName));
 
     if (rs != null) {
       try {
-        if (rs.next()) {
+        while (rs.next()) {
           String refColumn = rs.getString("FKCOLUMN_NAME");
-          result = refColumn != null && refColumn.equalsIgnoreCase(refColumnName);
+          if (refColumn != null && refColumn.equalsIgnoreCase(refColumnName)) {
+            result = true;
+          }
         }
       } finally {
         rs.close();
@@ -393,8 +406,15 @@ public class DBAccessorImpl implements DBAccessor {
     executeQuery(query, false);
   }
 
+  @Override
+  public ResultSet executeSelect(String query) throws SQLException {
+    Statement statement = getConnection().createStatement();
+    return statement.executeQuery(query);
+  }
+
   @Override
   public void executeQuery(String query, boolean ignoreFailure) throws SQLException {
+    LOG.info("Executing query: {}", query);
     Statement statement = getConnection().createStatement();
     try {
       statement.execute(query);
@@ -441,15 +461,20 @@ public class DBAccessorImpl implements DBAccessor {
    */
   public void executeScript(String filePath) throws SQLException, IOException {
     BufferedReader br = new BufferedReader(new FileReader(filePath));
-    ScriptRunner scriptRunner = new ScriptRunner(getConnection(), true, false);
+    ScriptRunner scriptRunner = new ScriptRunner(getConnection(), false, false);
     scriptRunner.runScript(br);
   }
 
   @Override
   public DatabaseSession getNewDatabaseSession() {
-    Login login = new DatabaseLogin();
+    DatabaseLogin login = new DatabaseLogin();
     login.setUserName(configuration.getDatabaseUser());
     login.setPassword(configuration.getDatabasePassword());
+    login.setDatasourcePlatform(databasePlatform);
+    login.setDatabaseURL(configuration.getDatabaseUrl());
+    login.setDriverClassName(configuration.getDatabaseDriver());
+
+
     return new DatabaseSessionImpl(login);
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/orm/helpers/ScriptRunner.java

@@ -216,7 +216,9 @@ public class ScriptRunner {
       printlnError(e);
       throw e;
     } finally {
-      conn.rollback();
+      if (!autoCommit) {
+        conn.rollback();
+      }
       flush();
     }
   }

ambari-server/src/main/java/org/apache/ambari/server/upgrade/AbstractUpgradeCatalog.java

@@ -99,17 +99,18 @@ public abstract class AbstractUpgradeCatalog implements UpgradeCatalog {
   /**
    * Update metainfo to new version.
    */
-  protected int updateMetaInfoVersion(String version) {
+  public int updateMetaInfoVersion(String version) {
+    //TODO verify version/server version usage
     String ambariServerVersion = getAmbariServerVersion();
     int rows = 0;
 
     if (ambariServerVersion != null) {
       try {
-        dbAccessor.executeQuery("INSERT INTO metainfo ('metainfo_key', " +
-          "'metainfo_value') VALUES ('version', '${ambariVersion}')");
+        dbAccessor.executeQuery(String.format("INSERT INTO metainfo (metainfo_key, " +
+          "metainfo_value) VALUES ('version', '%s')", version), true);
 
         rows = dbAccessor.updateTable("metainfo", "metainfo_value",
-          ambariServerVersion, String.format("WHERE metainfo_key = '%s'", version));
+          version, "WHERE metainfo_key = 'version'");
       } catch (SQLException e) {
         LOG.error("Failed updating metainfo table.", e);
       }
@@ -160,18 +161,16 @@ public abstract class AbstractUpgradeCatalog implements UpgradeCatalog {
     String dbUser = configuration.getDatabaseUser();
     String dbName = configuration.getServerDBName();
 
+    //wrap username with double quotes to accept old username "ambari-server"
+    if (!dbUser.contains("\"")) {
+      dbUser = String.format("\"%s\"", dbUser);
+    }
+
     dbAccessor.executeQuery(String.format("ALTER SCHEMA %s OWNER TO %s;", dbName, dbUser));
 
     dbAccessor.executeQuery(String.format("ALTER ROLE %s SET search_path to '%s';", dbUser, dbName));
   }
 
-  protected void grantAllPostgresPrivileges() throws SQLException {
-    String query = "GRANT ALL PRIVILEGES ON DATABASE ambari TO " +
-      configuration.getDatabaseUser();
-
-    dbAccessor.executeQuery(query);
-  }
-
   @Override
   public void upgradeSchema() throws AmbariException, SQLException {
     if (getDbType().equals(Configuration.POSTGRES_DB_NAME)) {

ambari-server/src/main/java/org/apache/ambari/server/upgrade/SchemaUpgradeHelper.java

@@ -25,11 +25,14 @@ import com.google.inject.persist.PersistService;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.controller.ControllerModule;
+import org.apache.ambari.server.orm.DBAccessor;
 import org.apache.ambari.server.orm.dao.MetainfoDAO;
 import org.apache.ambari.server.orm.entities.MetainfoEntity;
 import org.apache.ambari.server.utils.VersionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -42,15 +45,19 @@ public class SchemaUpgradeHelper {
     (SchemaUpgradeHelper.class);
 
   private Set<UpgradeCatalog> allUpgradeCatalogs;
-  private MetainfoDAO metainfoDAO;
   private PersistService persistService;
+  private DBAccessor dbAccessor;
+  private Configuration configuration;
 
   @Inject
-  public SchemaUpgradeHelper(Set<UpgradeCatalog> allUpgradeCatalogs, MetainfoDAO metainfoDAO,
-                             PersistService persistService) {
+  public SchemaUpgradeHelper(Set<UpgradeCatalog> allUpgradeCatalogs,
+                             PersistService persistService,
+                             DBAccessor dbAccessor,
+                             Configuration configuration) {
     this.allUpgradeCatalogs = allUpgradeCatalogs;
-    this.metainfoDAO = metainfoDAO;
     this.persistService = persistService;
+    this.dbAccessor = dbAccessor;
+    this.configuration = configuration;
   }
 
   private void startPersistenceService() {
@@ -68,17 +75,19 @@ public class SchemaUpgradeHelper {
   private String readSourceVersion() {
     String sourceVersion = null;
 
-    MetainfoEntity metainfoEntity = metainfoDAO.findByKey
-      (Configuration.SERVER_VERSION_KEY);
-
-    if (metainfoEntity != null) {
-      String version = metainfoEntity.getMetainfoValue();
-      if (version != null) {
-        sourceVersion = VersionUtils.getVersionSubstring(version);
+    try {
+      ResultSet resultSet = dbAccessor.executeSelect("SELECT metainfo_value from metainfo WHERE metainfo_key='version'");
+      if (resultSet.next()) {
+        return resultSet.getString(1);
+      } else {
+        //not found, assume oldest version
+        //doesn't matter as there single upgrade catalog for 1.2.0 - 1.5.0 and 1.4.4 - 1.5.0 upgrades
+        return "1.2.0";
       }
+    } catch (SQLException e) {
+      throw new RuntimeException("Unable to read database version", e);
     }
 
-    return sourceVersion;
   }
 
   /**

ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog150.java

@@ -223,15 +223,10 @@ public class UpgradeCatalog150 extends AbstractUpgradeCatalog {
     // Alter columns
 
     if (getDbType().equals(Configuration.POSTGRES_DB_NAME)) {
-      try {
-        dbAccessor.executeQuery("ALTER TABLE hostcomponentdesiredconfigmapping rename to hcdesiredconfigmapping;");
-        dbAccessor.executeQuery("ALTER TABLE users ALTER column user_id DROP DEFAULT;");
-        dbAccessor.executeQuery("ALTER TABLE users ALTER column ldap_user TYPE INTEGER USING CASE WHEN ldap_user=true THEN 1 ELSE 0 END;");
-        dbAccessor.executeQuery("ALTER TABLE hosts DROP COLUMN disks_info;");
-      } catch (SQLException e) {
-        LOG.warn("Error encountered while altering schema. ", e);
-        // continue updates
-      }
+      dbAccessor.executeQuery("ALTER TABLE hostcomponentdesiredconfigmapping rename to hcdesiredconfigmapping;", true);
+      dbAccessor.executeQuery("ALTER TABLE users ALTER column user_id DROP DEFAULT;", true);
+      dbAccessor.executeQuery("ALTER TABLE users ALTER column ldap_user TYPE INTEGER USING CASE WHEN ldap_user=true THEN 1 ELSE 0 END;", true);
+      dbAccessor.executeQuery("ALTER TABLE hosts DROP COLUMN disks_info;", true);
     }
 
     //Move tables from ambarirca db to ambari db; drop ambarirca; Mysql
@@ -259,8 +254,8 @@ public class UpgradeCatalog150 extends AbstractUpgradeCatalog {
     dbAccessor.addFKConstraint("configgroup", "FK_configgroup_cluster_id", "cluster_id", "clusters", "cluster_id", true);
     dbAccessor.addFKConstraint("confgroupclusterconfigmapping", "FK_cg_cluster_cm_config_tag", new String[] {"version_tag", "config_type", "cluster_id"}, "clusterconfig", new String[] {"version_tag", "type_name", "cluster_id"}, true);
     dbAccessor.addFKConstraint("confgroupclusterconfigmapping", "FK_cg_cluster_cm_group_id", "config_group_id", "configgroup", "group_id", true);
-    dbAccessor.addFKConstraint("confgrouphostmapping", "FK_cghostm_configgroup_id", "config_group_id", "configgroup", "group_id", true);
-    dbAccessor.addFKConstraint("confgrouphostmapping", "FK_cghostm_host_name", "host_name", "hosts", "host_name", true);
+    dbAccessor.addFKConstraint("configgrouphostmapping", "FK_cghostm_configgroup_id", "config_group_id", "configgroup", "group_id", true);
+    dbAccessor.addFKConstraint("configgrouphostmapping", "FK_cghostm_host_name", "host_name", "hosts", "host_name", true);
     dbAccessor.addFKConstraint("clusterconfigmapping", "FK_clustercfgmap_cluster_id", "cluster_id", "clusters", "cluster_id", true);
 
     // ========================================================================
@@ -436,18 +431,18 @@ public class UpgradeCatalog150 extends AbstractUpgradeCatalog {
       "SELECT 'host_role_command_id_seq', COALESCE((SELECT max(task_id) FROM host_role_command), 1) + 50 " +
       "UNION ALL " +
       "SELECT 'configgroup_id_seq', 1";
+    //TODO verify issue with configgroup_id_seq which was added later, possibly need to be added for mysql and oracle and separated for postgres
   }
 
   private String getPostgresRequestUpgradeQuery() {
-    return "insert into request" +
-      "(request_id, cluster_id, request_context, start_time, end_time, create_time) " +
-      "(select distinct s.request_id, s.cluster_id, s.request_context, " +
-      "coalesce (cmd.start_time, -1), coalesce (cmd.end_time, -1), -1 " +
-      "from " +
-      "(select distinct request_id, cluster_id, request_context from stage ) s " +
-      "left join " +
-      "(select request_id, min(start_time) as start_time, max(end_time) " +
-      "as end_time from host_role_command group by request_id) cmd";
+    return "insert into ambari.request(request_id, cluster_id, request_context, start_time, end_time, create_time) (\n" +
+      "  select distinct s.request_id, s.cluster_id, s.request_context, coalesce (cmd.start_time, -1), coalesce (cmd.end_time, -1), -1\n" +
+      "  from\n" +
+      "    (select distinct request_id, cluster_id, request_context from ambari.stage ) s\n" +
+      "    left join\n" +
+      "    (select request_id, min(start_time) as start_time, max(end_time) as end_time from ambari.host_role_command group by request_id) cmd\n" +
+      "    on s.request_id=cmd.request_id\n" +
+      ")";
   }
 
   private String getOracleRequestUpgradeQuery() {
@@ -490,10 +485,6 @@ public class UpgradeCatalog150 extends AbstractUpgradeCatalog {
       LOG.error("Error reading file.", e);
     }
 
-    // TODO: Verify if this is necessary and possible
-    if (dbType.equals(Configuration.POSTGRES_DB_NAME)) {
-      grantAllPostgresPrivileges();
-    }
   }
 
   @Override

ambari-server/src/main/python/ambari-server.py

@@ -178,12 +178,12 @@ SECURITY_PROVIDER_KEY_CMD="{0}" + os.sep + "bin" + os.sep + "java -cp {1}" +\
 SCHEMA_UPGRADE_HELPER_CMD="{0}" + os.sep + "bin" + os.sep + "java -cp {1}" +\
                           os.pathsep + "{2} " +\
                           "org.apache.ambari.server.upgrade.SchemaUpgradeHelper" +\
-                          " {3}"
+                          " {3} > " + SERVER_OUT_FILE + " 2>&1"
 
 STACK_UPGRADE_HELPER_CMD="{0}" + os.sep + "bin" + os.sep + "java -cp {1}" +\
                           os.pathsep + "{2} " +\
                           "org.apache.ambari.server.upgrade.StackUpgradeHelper" +\
-                          " {3} {4}"
+                          " {3} {4} > " + SERVER_OUT_FILE + " 2>&1"
 
 
 SECURITY_KEYS_DIR = "security.server.keys_dir"
@@ -239,6 +239,8 @@ SETUP_DB_CMD = ['su', '-', 'postgres',
 UPGRADE_STACK_CMD = ['su', 'postgres',
         '--command=psql -f {0} -v stack_name="\'{1}\'"  -v stack_version="\'{2}\'" -v dbname="{3}"']
 
+CHANGE_OWNER_COMMAND = ['su', '-', 'postgres',
+                        '--command=/var/lib/ambari-server/resources/scripts/change_owner.sh -d {0} -s {1} -o {2}']
 
 PG_ERROR_BLOCKED = "is being accessed by other users"
 PG_STATUS_RUNNING = "running"
@@ -263,6 +265,8 @@ POSTGRESQL_CONF_FILE = os.path.join(PG_HBA_DIR, "postgresql.conf")
 
 SERVER_VERSION_FILE_PATH = "server.version.file"
 
+#TODO property used incorrectly in local case, it was meant to be dbms name, not postgres database name,
+# has workaround for now, as we don't need dbms name if persistence_type=local
 JDBC_DATABASE_PROPERTY = "server.jdbc.database"
 JDBC_HOSTNAME_PROPERTY = "server.jdbc.hostname"
 JDBC_PORT_PROPERTY = "server.jdbc.port"
@@ -1127,7 +1131,7 @@ def get_pass_file_path(conf_file):
 
 # Set database properties to default values
 def load_default_db_properties(args):
-  args.database=DATABASE_NAMES[DATABASE_INDEX]
+  args.dbms=DATABASE_NAMES[DATABASE_INDEX]
   args.database_host = "localhost"
   args.database_port = DATABASE_PORTS[DATABASE_INDEX]
   args.database_name = DEFAULT_DB_NAME
@@ -1159,9 +1163,9 @@ def prompt_db_properties(args):
       )
 
       DATABASE_INDEX = int(database_num) - 1
-      args.database = DATABASE_NAMES[DATABASE_INDEX]
+      args.dbms = DATABASE_NAMES[DATABASE_INDEX]
       
-      if args.database != "postgres" :
+      if args.dbms != "postgres" :
         args.database_host = get_validated_string_input(
           "Hostname (" + args.database_host + "): ",
           args.database_host,
@@ -1179,7 +1183,7 @@ def prompt_db_properties(args):
           False
         )
 
-        if args.database == "oracle":
+        if args.dbms == "oracle":
           # Oracle uses service name or service id
           idType = "1"
           idType = get_validated_string_input(
@@ -1197,7 +1201,7 @@ def prompt_db_properties(args):
           IDTYPE_INDEX = int(idType) - 1
           args.database_name = get_validated_service_name(args.database_name, 
                                                           IDTYPE_INDEX)
-        elif args.database == "mysql":
+        elif args.dbms == "mysql":
           args.database_name = get_validated_db_name(args.database_name)
 
         else:
@@ -1223,7 +1227,7 @@ def prompt_db_properties(args):
       args.database_password =  configure_database_password(True)
 
   print_info_msg('Using database options: {database},{host},{port},{schema},{user},{password}'.format(
-    database=args.database,
+    database=args.dbms,
     host=args.database_host,
     port=args.database_port,
     schema=args.database_name,
@@ -1243,7 +1247,7 @@ def store_remote_properties(args):
 
   properties.process_pair(PERSISTENCE_TYPE_PROPERTY, "remote")
 
-  properties.process_pair(JDBC_DATABASE_PROPERTY, args.database)
+  properties.process_pair(JDBC_DATABASE_PROPERTY, args.dbms)
   properties.process_pair(JDBC_HOSTNAME_PROPERTY, args.database_host)
   properties.process_pair(JDBC_PORT_PROPERTY, args.database_port)
   properties.process_pair(JDBC_SCHEMA_PROPERTY, args.database_name)
@@ -1339,7 +1343,7 @@ def execute_remote_script(args, scriptPath):
       print_warning_msg('{0} not found'.format(DATABASE_CLI_TOOLS[DATABASE_INDEX]))
     return -1, "Client wasn't found", "Client wasn't found"
 
-  if args.database == "postgres":
+  if args.dbms == "postgres":
 
     os.environ["PGPASSWORD"] = args.database_password
     retcode, out, err = run_in_shell('{0} {1}'.format(tool,  POSTGRES_EXEC_ARGS.format(
@@ -1350,7 +1354,7 @@ def execute_remote_script(args, scriptPath):
       scriptPath
     )))
     return retcode, out, err
-  elif args.database == "oracle":
+  elif args.dbms == "oracle":
     sid_or_sname = "sid"
     if (hasattr(args, 'sid_or_sname') and args.sid_or_sname == "sname") or \
         (hasattr(args, 'jdbc_url') and args.jdbc_url and re.match(ORACLE_SNAME_PATTERN, args.jdbc_url)):
@@ -1367,7 +1371,7 @@ def execute_remote_script(args, scriptPath):
       sid_or_sname
     )))
     return retcode, out, err
-  elif args.database == "mysql":
+  elif args.dbms == "mysql":
     MYSQL_EXEC_ARGS = MYSQL_EXEC_ARGS_WO_USER_VARS if MYSQL_INIT_SCRIPT == scriptPath else MYSQL_EXEC_ARGS_WITH_USER_VARS
     retcode, out, err = run_in_shell('{0} {1}'.format(tool, MYSQL_EXEC_ARGS.format(
       args.database_host,
@@ -1381,6 +1385,7 @@ def execute_remote_script(args, scriptPath):
 
   return -2, "Wrong database", "Wrong database"
 
+
 def configure_database_password(showDefault=True):
   passwordDefault = PG_DEFAULT_PASSWORD
   if showDefault:
@@ -1396,10 +1401,11 @@ def configure_database_password(showDefault=True):
 
   return password
 
+
 def check_database_name_property():
   properties = get_ambari_properties()
   if properties == -1:
-    print_error_msg ("Error getting ambari properties")
+    print_error_msg("Error getting ambari properties")
     return -1
 
   dbname = properties[JDBC_DATABASE_PROPERTY]
@@ -1432,6 +1438,7 @@ def configure_database_username_password(args):
   else:
     print_error_msg("Connection properties not set in config file.")
 
+
 # Check if jdbc user is changed
 def is_jdbc_user_changed(args):
   properties = get_ambari_properties()
@@ -1450,6 +1457,7 @@ def is_jdbc_user_changed(args):
 
   return None
 
+
 # Store local database connection properties
 def store_local_properties(args):
   properties = get_ambari_properties()
@@ -1518,13 +1526,13 @@ def parse_properties_file(args):
     args.persistence_type = "local"
 
   if args.persistence_type == 'remote':
-    args.database = properties[JDBC_DATABASE_PROPERTY]
+    args.dbms = properties[JDBC_DATABASE_PROPERTY]
     args.database_host = properties[JDBC_HOSTNAME_PROPERTY]
     args.database_port = properties[JDBC_PORT_PROPERTY]
     args.database_name = properties[JDBC_SCHEMA_PROPERTY]
     global DATABASE_INDEX
     try:
-      DATABASE_INDEX = DATABASE_NAMES.index(args.database)
+      DATABASE_INDEX = DATABASE_NAMES.index(args.dbms)
     except ValueError:
       pass
   else:
@@ -2010,7 +2018,7 @@ def find_jdk():
 # Checks if options determine local DB configuration
 #
 def is_local_database(options):
-  if options.database == DATABASE_NAMES[0] \
+  if options.dbms == DATABASE_NAMES[0] \
     and options.database_host == "localhost" \
     and options.database_port == DATABASE_PORTS[0]:
     return True
@@ -2018,9 +2026,9 @@ def is_local_database(options):
 
 #Check if required jdbc drivers present
 def find_jdbc_driver(args):
-  if args.database in JDBC_PATTERNS.keys():
+  if args.dbms in JDBC_PATTERNS.keys():
     drivers = []
-    drivers.extend(glob.glob(JAVA_SHARE_PATH + os.sep + JDBC_PATTERNS[args.database]))
+    drivers.extend(glob.glob(JAVA_SHARE_PATH + os.sep + JDBC_PATTERNS[args.dbms]))
     if drivers:
       return drivers
     return -1
@@ -2062,7 +2070,7 @@ def check_jdbc_drivers(args):
   
   msg = 'Before starting Ambari Server, ' \
         'you must copy the {0} JDBC driver JAR file to {1}.'.format(
-        DATABASE_FULL_NAMES[args.database],
+        DATABASE_FULL_NAMES[args.dbms],
         JAVA_SHARE_PATH)
 
   
@@ -2352,7 +2360,7 @@ def start(args):
     result = find_jdbc_driver(args)
     msg = 'Before starting Ambari Server, ' \
           'you must copy the {0} JDBC driver JAR file to {1}.'.format(
-          DATABASE_FULL_NAMES[args.database],
+          DATABASE_FULL_NAMES[args.dbms],
           JAVA_SHARE_PATH)
     if result == -1:
       raise FatalException(-1, msg)
@@ -2563,6 +2571,27 @@ def upgrade_local_repo(args):
     run_metainfo_upgrade(metainfo_update_items)
 
 
+def change_objects_owner(args):
+  print 'Fixing database objects owner'
+  database_name = args.database_name
+  new_owner = args.database_username
+  if '"' not in new_owner:
+    #wrap to allow old username "ambari-server", postgres only
+    new_owner = '\'"{0}"\''.format(new_owner)
+    pass
+
+  command = CHANGE_OWNER_COMMAND[:]
+  command[-1] = command[-1].format(database_name, 'ambari', new_owner)
+  return run_os_command(command)
+
+
+def compare_versions(version1, version2):
+  def normalize(v):
+    return [int(x) for x in re.sub(r'(\.0+)*$', '', v).split(".")]
+  return cmp(normalize(version1), normalize(version2))
+  pass
+
+
 def run_schema_upgrade(version):
   jdk_path = find_jdk()
   if jdk_path is None:
@@ -2570,8 +2599,7 @@ def run_schema_upgrade(version):
                     "command to install a JDK automatically or install any "
                     "JDK manually to " + JDK_INSTALL_DIR)
     return 1
-  command = SCHEMA_UPGRADE_HELPER_CMD.format(jdk_path, get_ambari_classpath(),
-                                             get_conf_dir(), version)
+  command = SCHEMA_UPGRADE_HELPER_CMD.format(jdk_path, get_conf_dir(), get_ambari_classpath(), version)
   (retcode, stdout, stderr) = run_os_command(command)
   print_info_msg("Return code from schema upgrade command, retcode = " + str(retcode))
   if retcode > 0:
@@ -2588,8 +2616,8 @@ def run_stack_upgrade(stackName, stackVersion):
   stackId = {}
   stackId[stackName] = stackVersion
 
-  command = STACK_UPGRADE_HELPER_CMD.format(jdk_path, get_ambari_classpath(),
-                                            get_conf_dir(), "updateStackId",
+  command = STACK_UPGRADE_HELPER_CMD.format(jdk_path, get_conf_dir(), get_ambari_classpath(),
+                                             "updateStackId",
                                             json.dumps(stackId))
   (retcode, stdout, stderr) = run_os_command(command)
   print_info_msg("Return code from stack upgrade command, retcode = " + str(retcode))
@@ -2606,8 +2634,8 @@ def run_metainfo_upgrade(keyValueMap = None):
 
   retcode = 1
   if keyValueMap:
-    command = STACK_UPGRADE_HELPER_CMD.format(jdk_path, get_ambari_classpath(),
-                                              get_conf_dir(), 'updateMetaInfo',
+    command = STACK_UPGRADE_HELPER_CMD.format(jdk_path, get_conf_dir(), get_ambari_classpath(),
+                                              'updateMetaInfo',
                                               json.dumps(keyValueMap))
     (retcode, stdout, stderr) = run_os_command(command)
     print_info_msg("Return code from stack upgrade command, retcode = " + str(retcode))
@@ -2657,12 +2685,19 @@ def upgrade(args):
       server_version = f.read()
 
   if not server_version:
-    raise FatalException('Cannot determine server version from version file '
+    raise FatalException(10, 'Cannot determine server version from version file '
                          '%s' % args.server_version_file_path)
 
+  #fix local database objects owner in pre 1.5.0
+  #TODO check database version
+  if args.persistence_type == 'local':
+    retcode, stdout, stderr = change_objects_owner(args)
+    if not retcode == 0:
+      raise FatalException(20, 'Unable to change owner of database objects')
+
   retcode = run_schema_upgrade(server_version.strip())
   if not retcode == 0:
-    raise FatalException('Scehma upgrade failed.')
+    raise FatalException(11, 'Schema upgrade failed.')
 
   user = read_ambari_user()
   if user is None:
@@ -3923,7 +3958,7 @@ def main():
   parser.add_option('-g', '--debug', action="store_true", dest='debug', default=False,
                     help="Start ambari-server in debug mode")
 
-  parser.add_option('--database', default=None, help ="Database to use postgres|oracle", dest="database")
+  parser.add_option('--database', default=None, help ="Database to use postgres|oracle", dest="dbms")
   parser.add_option('--databasehost', default=None, help="Hostname of database server", dest="database_host")
   parser.add_option('--databaseport', default=None, help="Database port", dest="database_port")
   parser.add_option('--databasename', default=None, help="Database/Schema/Service name or ServiceID",
@@ -3952,7 +3987,7 @@ def main():
 
   options.warnings = []
 
-  if options.database is None \
+  if options.dbms is None \
     and options.database_host is None \
     and options.database_port is None \
     and options.database_name is None \
@@ -3961,7 +3996,7 @@ def main():
 
     PROMPT_DATABASE_OPTIONS = True
 
-  elif not (options.database is not None
+  elif not (options.dbms is not None
     and options.database_host is not None
     and options.database_port is not None
     and options.database_name is not None
@@ -3970,12 +4005,12 @@ def main():
     parser.error('All database options should be set. Please see help for the options.')
 
   #correct database
-  if options.database is not None and options.database not in DATABASE_NAMES:
+  if options.dbms is not None and options.dbms not in DATABASE_NAMES:
     parser.print_help()
-    parser.error("Unsupported Database " + options.database)
-  elif options.database is not None:
-    options.database = options.database.lower()
-    DATABASE_INDEX = DATABASE_NAMES.index(options.database)
+    parser.error("Unsupported Database " + options.dbms)
+  elif options.dbms is not None:
+    options.dbms = options.dbms.lower()
+    DATABASE_INDEX = DATABASE_NAMES.index(options.dbms)
 
   #correct port
   if options.database_port is not None:
@@ -3990,7 +4025,7 @@ def main():
       parser.print_help()
       parser.error("Incorrect database port " + options.database_port)
 
-  if options.database is not None and options.database == "postgres":
+  if options.dbms is not None and options.dbms == "postgres":
     print "WARNING: HostName for postgres server " + options.database_host + \
      " will be ignored: using localhost."
     options.database_host = "localhost"

ambari-server/src/main/resources/scripts/change_owner.sh

@@ -0,0 +1,66 @@
+#!/bin/bash
+# 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.
+
+usage()
+{
+cat << EOF
+usage: $0 options
+
+This script set ownership for all table, sequence and views for a given database
+
+OPTIONS:
+   -h      Show this message
+   -d      Database name
+   -o      Owner
+   -s      Schema (defaults to public)
+EOF
+}
+
+DB_NAME="";
+NEW_OWNER="";
+SCHEMA="public";
+while getopts "hd:o:s:" OPTION; do
+    case $OPTION in
+        h)
+            usage;
+            exit 1;
+            ;;
+        d)
+            DB_NAME=$OPTARG;
+            ;;
+        o)
+            NEW_OWNER=$OPTARG;
+            ;;
+        s)
+            SCHEMA=$OPTARG;
+            ;;
+    esac
+done
+
+if [[ -z $DB_NAME ]] || [[ -z $NEW_OWNER ]]; then
+     usage;
+     exit 1;
+fi
+
+# Using the NULL byte as the separator as its the only character disallowed from PG table names
+IFS=\0;
+for tbl in `psql -qAt -R\0 -c "SELECT tablename FROM pg_tables WHERE schemaname = '${SCHEMA}';" ${DB_NAME}` \
+           `psql -qAt -R\0 -c "SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = '${SCHEMA}';" ${DB_NAME}` \
+           `psql -qAt -R\0 -c "SELECT table_name FROM information_schema.views WHERE table_schema = '${SCHEMA}';" ${DB_NAME}` ;
+do
+    psql -c "ALTER TABLE \"${SCHEMA}\".\"$tbl\" OWNER TO ${NEW_OWNER}" ${DB_NAME};
+done
+unset IFS;

ambari-server/src/test/java/org/apache/ambari/server/orm/DBAccessorImplTest.java

@@ -21,6 +21,7 @@ package org.apache.ambari.server.orm;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import junit.framework.Assert;
+import org.eclipse.persistence.sessions.DatabaseSession;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
@@ -34,6 +35,8 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Vector;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.apache.ambari.server.orm.DBAccessor.DBColumnInfo;
@@ -270,4 +273,38 @@ public class DBAccessorImplTest {
 
   }
 
+  @Test
+  public void testExecuteSelect() throws Exception {
+    DBAccessorImpl dbAccessor = injector.getInstance(DBAccessorImpl.class);
+    String tableName = getFreeTableName();
+    createMyTable(tableName);
+    dbAccessor.executeQuery("insert into " + tableName + "(id, name, time) values(1, 'Bob', 1234567)");
+
+    ResultSet resultSet = dbAccessor.executeSelect("select name from " + tableName + " where id=1");
+    int count = 0;
+    while (resultSet.next()) {
+      assertEquals("Bob", resultSet.getString(1));
+      count++;
+    }
+
+    assertEquals(count, 1);
+  }
+
+  @Test
+  public void testDBSession() throws Exception {
+    DBAccessorImpl dbAccessor = injector.getInstance(DBAccessorImpl.class);
+    String tableName = getFreeTableName();
+    createMyTable(tableName);
+    dbAccessor.executeQuery("insert into " + tableName + "(id, name, time) values(1, 'Bob', 1234567)");
+
+    DatabaseSession databaseSession = dbAccessor.getNewDatabaseSession();
+    databaseSession.login();
+    Vector vector = databaseSession.executeSQL("select * from " + tableName + " where id=1");
+    assertEquals(vector.size(), 1);
+    Map map = (Map) vector.get(0);
+    //all names seem to be converted to upper case
+    assertEquals("Bob", map.get("name".toUpperCase()));
+
+    databaseSession.logout();
+  }
 }

ambari-server/src/test/python/TestAmbariServer.py

@@ -212,7 +212,7 @@ class TestAmbariServer(TestCase):
     options = MagicMock()
     args = ["setup-security"]
     opm.parse_args.return_value = (options, args)
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sid"
     setup_security_method.return_value = None
 
@@ -306,7 +306,7 @@ class TestAmbariServer(TestCase):
     args = ["setup"]
     opm.parse_args.return_value = (options, args)
 
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sid"
     ambari_server.main()
 
@@ -331,7 +331,7 @@ class TestAmbariServer(TestCase):
     args = ["setup"]
     opm.parse_args.return_value = (options, args)
 
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sname"
     ambari_server.main()
 
@@ -356,7 +356,7 @@ class TestAmbariServer(TestCase):
     args = ["start", "-g"]
     opm.parse_args.return_value = (options, args)
 
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sid"
 
     ambari_server.main()
@@ -380,7 +380,7 @@ class TestAmbariServer(TestCase):
     options = MagicMock()
     args = ["start", "--debug"]
     opm.parse_args.return_value = (options, args)
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sid"
 
     ambari_server.main()
@@ -405,7 +405,7 @@ class TestAmbariServer(TestCase):
     args = ["stop"]
     opm.parse_args.return_value = (options, args)
 
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sid"
 
     ambari_server.main()
@@ -431,7 +431,7 @@ class TestAmbariServer(TestCase):
     options = MagicMock()
     args = ["reset"]
     opm.parse_args.return_value = (options, args)
-    options.database = None
+    options.dbms = None
     options.sid_or_sname = "sid"
 
     ambari_server.main()
@@ -2082,7 +2082,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     args = MagicMock()
     ambari_server.load_default_db_properties(args)
     ambari_server.prompt_db_properties(args)
-    self.assertEqual(args.database, "postgres")
+    self.assertEqual(args.dbms, "postgres")
     self.assertEqual(args.database_host, "localhost")
     self.assertEqual(args.database_name, "ambari")
     self.assertEqual(args.database_port, "5432")
@@ -2106,7 +2106,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     args = MagicMock()
     ambari_server.load_default_db_properties(args)
     ambari_server.prompt_db_properties(args)
-    self.assertEqual(args.database, "oracle")
+    self.assertEqual(args.dbms, "oracle")
     self.assertEqual(args.database_port, "1521")
     self.assertEqual(args.database_host, "localhost")
     self.assertEqual(args.database_name, "ambari")
@@ -2132,7 +2132,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     args = MagicMock()
     ambari_server.load_default_db_properties(args)
     ambari_server.prompt_db_properties(args)
-    self.assertEqual(args.database, "oracle")
+    self.assertEqual(args.dbms, "oracle")
     self.assertEqual(args.database_port, "1521")
     self.assertEqual(args.database_host, "localhost")
     self.assertEqual(args.database_name, "ambari")
@@ -2158,7 +2158,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     args = MagicMock()
     ambari_server.load_default_db_properties(args)
     ambari_server.prompt_db_properties(args)
-    self.assertEqual(args.database, "postgres")
+    self.assertEqual(args.dbms, "postgres")
     self.assertEqual(args.database_port, "5432")
     self.assertEqual(args.database_host, "localhost")
     self.assertEqual(args.database_name, "ambari")
@@ -2542,7 +2542,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
 
     # Remote DB
     args.persistence_type = "remote"
-    args.database = "oracle"
+    args.dbms = "oracle"
 
     # Case when jdbc driver is not used
     find_jdbc_driver_mock.return_value = -1
@@ -2760,9 +2760,9 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     self.assertTrue(get_ambari_classpath_mock.called)
     self.assertTrue(get_conf_dir_mock.called)
     self.assertTrue(run_os_command_mock.called)
-    run_os_command_mock.assert_called_with('/usr/lib/java/bin/java -cp '
-        'test:path12:/etc/conf org.apache.ambari.server.upgrade.StackUpgradeHelper '
-        'updateStackId ' + json.dumps(stackIdMap))
+    run_os_command_mock.assert_called_with('/usr/lib/java/bin/java -cp /etc/conf:test:path12 '
+                                           'org.apache.ambari.server.upgrade.StackUpgradeHelper '
+        'updateStackId ' + json.dumps(stackIdMap) + ' > /var/log/ambari-server/ambari-server.out 2>&1')
 
 
   @patch.object(ambari_server, 'get_conf_dir')
@@ -2782,9 +2782,9 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     self.assertTrue(get_ambari_classpath_mock.called)
     self.assertTrue(get_conf_dir_mock.called)
     self.assertTrue(run_os_command_mock.called)
-    run_os_command_mock.assert_called_with('/usr/lib/java/bin/java -cp '
-        'test:path12:/etc/conf org.apache.ambari.server.upgrade.SchemaUpgradeHelper '
-        '1.4.9.40')
+    run_os_command_mock.assert_called_with('/usr/lib/java/bin/java -cp /etc/conf:test:path12 '
+                                           'org.apache.ambari.server.upgrade.SchemaUpgradeHelper 1.4.9.40 '
+                                           '> /var/log/ambari-server/ambari-server.out 2>&1')
 
 
   @patch.object(ambari_server, 'get_conf_dir')
@@ -2805,9 +2805,9 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     self.assertTrue(get_ambari_classpath_mock.called)
     self.assertTrue(get_conf_dir_mock.called)
     self.assertTrue(run_os_command_mock.called)
-    run_os_command_mock.assert_called_with('/usr/lib/java/bin/java -cp '
-        'test:path12:/etc/conf org.apache.ambari.server.upgrade.StackUpgradeHelper '
-        'updateMetaInfo ' + json.dumps(json_map))
+    run_os_command_mock.assert_called_with('/usr/lib/java/bin/java -cp /etc/conf:test:path12 '
+                                           'org.apache.ambari.server.upgrade.StackUpgradeHelper updateMetaInfo ' +
+                                           json.dumps(json_map) + ' > /var/log/ambari-server/ambari-server.out 2>&1')
 
 
   @patch("__builtin__.open")
@@ -2923,7 +2923,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
 
     ambari_server.load_default_db_properties(args)
 
-    self.assertEquals(args.database, "postgres")
+    self.assertEquals(args.dbms, "postgres")
     self.assertEquals(args.database_host, "localhost")
     self.assertEquals(args.database_port, "5432")
     self.assertEquals(args.database_name, "ambari")
@@ -2933,7 +2933,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     args = MagicMock()
     ambari_server.DATABASE_INDEX = 1
     ambari_server.load_default_db_properties(args)
-    self.assertEquals(args.database, "oracle")
+    self.assertEquals(args.dbms, "oracle")
     self.assertEquals(args.database_port, "1521")
 
     failed = False
@@ -3082,7 +3082,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     self.assertTrue(get_validated_string_input_mock.called)
     self.assertTrue(load_default_db_properties_mock.called)
 
-    self.assertEquals(args.database, "postgres")
+    self.assertEquals(args.dbms, "postgres")
 
     load_default_db_properties_mock.reset_mock()
     get_validated_string_input_mock.reset_mock()
@@ -3101,7 +3101,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     #success
     run_in_shell_mock.return_value = (0, None, None)
     get_db_cli_tool_mock.return_value = "tool"
-    args.database = "postgres"
+    args.dbms = "postgres"
     args.database_password = "some_password"
 
     retcode, out, err = ambari_server.execute_remote_script(args, script)
@@ -3136,7 +3136,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     run_in_shell_mock.return_value = (0, None, None)
     get_db_cli_tool_mock.called = False
     run_in_shell_mock.called = False
-    args.database = "unknown"
+    args.dbms = "unknown"
 
     retcode, out, err = ambari_server.execute_remote_script(args, script)
     self.assertEquals(retcode, -2)
@@ -3341,7 +3341,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     import optparse
 
     args = optparse.Values()
-    args.database = "oracle"
+    args.dbms = "oracle"
     args.database_host = "localhost"
     args.database_port = "1234"
     args.database_name = "ambari"
@@ -3446,7 +3446,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     get_ambari_properties_mock.return_value = {ambari_server.RESOURCES_DIR_PROPERTY: resources_dir}
     find_jdbc_driver_mock.return_value = drivers_list
 
-    args.database = "oracle"
+    args.dbms = "oracle"
 
     rcode = ambari_server.check_jdbc_drivers(args)
 
@@ -3582,7 +3582,7 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     find_jdbc_driver_mock.return_value = drivers_list
     copy_files_mock.return_value = 0
 
-    args.database = "oracle"
+    args.dbms = "oracle"
 
     rcode = ambari_server.check_jdbc_drivers(args)