CI: add gpMgmt Behave test

gpAux/gpdemo/demo_cluster.sh

@@ -314,8 +314,10 @@ cat >> $CLUSTER_CONFIG <<-EOF
 
 	COORDINATOR_PORT=${COORDINATOR_DEMO_PORT}
 
-	# Shell to use to execute commands on all hosts
-	TRUSTED_SHELL="$(dirname "$0")/lalshell"
+	# Shell to use to execute commands on all hosts. Use an absolute path here
+	# because this file is later sourced by gpinitsystem, where \$0 is no longer
+	# demo_cluster.sh.
+	TRUSTED_SHELL=$(pwd)/lalshell
 
 	ENCODING=UNICODE
 EOF

gpMgmt/bin/analyzedb

@@ -982,7 +982,11 @@ def get_oid_str(table_list):
 def regclass_schema_tbl(schema, tbl):
     schema_tbl = "%s.%s" % (escape_identifier(schema), escape_identifier(tbl))
 
-    return "to_regclass('%s')" % (pg.escape_string(schema_tbl))
+    # Not pg.escape_string(): PyGreSQL's C implementation encodes its argument
+    # as ASCII and raises UnicodeEncodeError on a table or schema name that
+    # contains non-ASCII characters. With standard_conforming_strings on -- the
+    # default -- doubling single quotes is the whole of the escaping needed.
+    return "to_regclass('%s')" % schema_tbl.replace("'", "''")
 
 
 # Escape double-quotes in a string, so that the resulting string is suitable for

gpMgmt/bin/gpcheckcat

@@ -1075,9 +1075,12 @@ def checkOwners():
                     a.rolname, m.rolname as coordinator_rolname
     from gp_dist_random('pg_class') r
       join pg_class c on (c.oid = r.oid)
+      left join pg_index i on (c.oid = i.indexrelid)
       left join pg_appendonly ao on (c.oid = ao.segrelid or
                                      c.oid = ao.blkdirrelid or
-                                     c.oid = ao.blkdiridxid)
+                                     c.oid = ao.visimaprelid or
+                                     i.indrelid = ao.blkdirrelid or
+                                     i.indrelid = ao.visimaprelid)
       left join pg_class o on (o.oid = ao.relid or
                                o.reltoastrelid = c.oid)
       join pg_authid a on (a.oid = r.relowner)
@@ -1948,6 +1951,203 @@ def checkOrphanedToastTables():
                                issue_type="orphaned_toast_tables",
                                description='Repairing orphaned TOAST tables')
 
+def fetch_guc_value(guc):
+    qry = '''
+    show {}
+     '''.format(guc)
+    try:
+        conn = connect2(GV.cfg[GV.coordinator_dbid])
+        curs = conn.query(qry)
+        rows = curs.getresult()
+        guc_value = rows[0][0]
+        return guc_value
+
+    except Exception as e:
+        setError(ERROR_NOREPAIR)
+        GV.checkStatus = False
+        myprint('[ERROR] executing test: mix_distribution_policy')
+        myprint('  Execution error: ' + str(e))
+
+def generateDistPolicyQueryFile():
+
+    query_sql = '''
+     -- all tables that use legacy policy:
+     with legacy_opclass_oids(oid_array) as (
+       select
+         array_agg(oid)
+       from
+         pg_opclass
+       where
+         opcfamily in (
+           select
+             amprocfamily
+           from
+             pg_amproc
+           where
+             amproc :: oid in (
+               6140, 6141, 6142, 6143, 6144, 6145, 6146,
+               6147, 6148, 6149, 6150, 6151, 6152,
+               6153, 6154, 6155, 6156, 6157, 6158,
+               6159, 6160, 6161, 6162, 6163, 6164,
+               6165, 6166, 6167, 6168, 6170, 6169,
+               6171
+             )
+         )
+     )
+    select
+         localoid :: regclass :: text as "Legacy Policy"
+    from
+         gp_distribution_policy,
+         legacy_opclass_oids
+     where
+         policytype = 'p'
+         and distclass :: oid[] && oid_array;
+
+ -- all tables that don't use any legacy policy:
+ with legacy_opclass_oids(oid_array) as (
+   select
+     array_agg(oid)
+   from
+     pg_opclass
+   where
+     opcfamily in (
+       select
+         amprocfamily
+       from
+         pg_amproc
+       where
+         amproc :: oid in (
+           6140, 6141, 6142, 6143, 6144, 6145, 6146,
+           6147, 6148, 6149, 6150, 6151, 6152,
+           6153, 6154, 6155, 6156, 6157, 6158,
+           6159, 6160, 6161, 6162, 6163, 6164,
+           6165, 6166, 6167, 6168, 6170, 6169,
+           6171
+         )
+     )
+ )
+select
+   localoid :: regclass :: text as "Non Legacy Policy"
+from
+   gp_distribution_policy,
+   legacy_opclass_oids
+where
+   policytype = 'p'
+   and not (distclass :: oid[] && oid_array);
+           '''
+    filename = 'gpcheckcat.distpolicy.sql'
+
+    if not os.path.exists(filename) :
+        try:
+            with open(filename, 'w') as fp:
+                fp.write(query_sql + "\n")
+        except Exception as e:
+            logger.warning('Unable to generate verify file for {}'.format(filename))
+
+
+# Test to check if there are tables that use both legacy opclass/non legacy opclass
+# in distribution policy
+def checkMixDistPolicy() :
+
+    qry = '''
+        with legacy_opclass_oids(oid_array) as (
+        select
+          array_agg(oid)
+        from
+          pg_opclass
+        where
+          opcfamily in (
+            select
+              amprocfamily
+            from
+              pg_amproc
+            where
+              amproc :: oid in (
+                6140, 6141, 6142, 6143, 6144, 6145, 6146,
+                6147, 6148, 6149, 6150, 6151, 6152,
+                6153, 6154, 6155, 6156, 6157, 6158,
+                6159, 6160, 6161, 6162, 6163, 6164,
+                6165, 6166, 6167, 6168, 6170, 6169,
+                6171
+            )
+        )
+       ),
+       all_hash_ops(dc) as (
+        select
+          distinct unnest(distclass :: oid[]) 
+        from
+          gp_distribution_policy
+        )
+        select
+            count(1) filter(
+        where
+            array[x.dc] && oid_array
+            ) as n_legacy_dist_class,
+            count(1) as n_total_dist_class
+        from
+        all_hash_ops x,
+        legacy_opclass_oids y;
+    '''
+
+    try:
+        conn = connect2(GV.cfg[GV.coordinator_dbid])
+        curs = conn.query(qry)
+        rows = curs.getresult()
+
+        if rows:
+            row = rows[0]
+            n_legacy_dist_class = row[0]
+            n_total_dist_class = row[1]
+            GV.checkStatus = False
+
+            if n_legacy_dist_class > 0 and n_total_dist_class > n_legacy_dist_class :
+                generateDistPolicyQueryFile()
+                #if this condition is true then we have mix distribution Policy
+                myprint(
+                    '[ERROR]: Found tables created using both legacy and non legacy hashops'
+                    ' in distribution policy.'
+                    'Please run the gpcheckcat.distpolicy.sql file to list the tables.'
+                )
+            else:
+                if (n_legacy_dist_class == 0 or n_legacy_dist_class == n_total_dist_class):
+                #if this condition is true then we dont have mix distribution policy
+                    gp_use_legacy_hashops = fetch_guc_value("gp_use_legacy_hashops")
+                    printDistPolicyMsg(gp_use_legacy_hashops,
+                                     n_legacy_dist_class,
+                                     n_total_dist_class
+                                     )
+
+    except Exception as e:
+        setError(ERROR_NOREPAIR)
+        GV.checkStatus = False
+        myprint('[ERROR] executing test: mix_distribution_policy')
+        myprint('  Execution error: ' + str(e))
+
+def printDistPolicyMsg(gp_use_legacy_hashops,n_legacy_dist_class, n_total_dist_class):
+
+    GV.checkStatus = True
+
+    if n_total_dist_class - n_legacy_dist_class > 0  and gp_use_legacy_hashops == "on":
+        myprint(
+            '[ERROR]: GUC gp_use_legacy_hashops is on.'
+            ' all newly created tables will use legacy hash ops by default for hash distributed table, '
+            'but there are tables using non-legacy hash ops in the cluster. '
+            'Please run the gpcheckcat.distpolicy.sql file to list the tables.'
+            )
+        GV.checkStatus = False
+
+    elif n_legacy_dist_class == 0 and gp_use_legacy_hashops == "off":
+        GV.checkStatus = True
+
+    elif n_legacy_dist_class > 0 and gp_use_legacy_hashops == "off" :
+        myprint(
+            '[ERROR]: GUC gp_use_legacy_hashops is off.'
+            ' all newly created tables will use non legacy hash ops by default for hash distributed table, '
+            'but there are tables using legacy hash ops in the cluster. '
+            'Please run the gpcheckcat.distpolicy.sql file to list the tables.'
+            )
+        GV.checkStatus = False
+
 
 ############################################################################
 # Help populating repair part for all checked types
@@ -2082,7 +2282,16 @@ all_checks = {
             "version": 'main',
             "order": 15,
             "online": False
+        },
+     "mix_distribution_policy":
+        {
+            "description": "Check for tables that use legacy opclass in distribution policy",
+            "fn": lambda: checkMixDistPolicy(),
+            "version": 'main',
+            "order": 17,
+            "online": True
         }
+
 }
 
 
@@ -2366,6 +2575,7 @@ TableMainColumn['pg_type_encoding'] = ['typid', 'pg_type']
 TableMainColumn['pg_window'] = ['winfnoid', 'pg_proc']
 TableMainColumn['pg_password_history'] = ['passhistroleid', 'pg_authid']
 TableMainColumn['pg_description'] = ['objoid', 'pg_description']
+TableMainColumn['pg_shdescription'] = ['objoid', 'pg_shdescription']
 
 # Table with OID (special case), these OIDs are known to be inconsistent
 TableMainColumn['pg_attrdef'] = ['adrelid', 'pg_class']

gpMgmt/bin/gpinitsystem

@@ -313,7 +313,7 @@ CHK_PARAMS () {
 		fi
 
 		if [ x"$ETCD_HOST_CONFIG" = x"" ] || [ x"$CLUSTER_BOOT_MODE" = x"DEMO" ]; then
-			LOG_MSG "[WARN]:-No ETCD cluster host config provided, use default configuration."
+			LOG_MSG "[INFO]:-No ETCD cluster host config provided, use default configuration."
 		else
 			ETCD_HOST_MACHINE_LIST=(`$CAT $ETCD_HOST_CONFIG`)
 			ETCD_HOST_CONFIG_NUM=${#ETCD_HOST_MACHINE_LIST[@]}
@@ -323,7 +323,7 @@ CHK_PARAMS () {
 		fi
 
 		if [ x"$FTS_HOST_CONFIG" = x"" ] || [ x"$CLUSTER_BOOT_MODE" = x"DEMO" ]; then
-			LOG_MSG "[WARN]:-No FTS cluster host config provided, use default configuration."
+			LOG_MSG "[INFO]:-No FTS cluster host config provided, use default configuration."
 		else
 			FTS_HOST_MACHINE_LIST=(`$CAT $FTS_HOST_CONFIG`)
 			FTS_HOST_CONFIG_NUM=${#FTS_HOST_MACHINE_LIST[@]}
@@ -1641,6 +1641,12 @@ FORCE_FTS_PROBE () {
         if [ x"$RESULT" == x"f" ]; then
         	break
         fi
+
+        # Ask FTS to probe now rather than waiting out gp_fts_probe_interval;
+        # without this the loop just re-reads an unchanged
+        # gp_segment_configuration sixty times in a couple of seconds and gives
+        # up before FTS has looked at the segments even once.
+        $PSQL -p $GP_PORT -d "$DEFAULTDB" -c "select gp_request_fts_probe_scan()" >> ${LOG_FILE} 2>&1
     done
     LOG_MSG "[INFO]:-End Function $FUNCNAME"
 }

gpMgmt/bin/gppylib/mainUtils.py

@@ -307,13 +307,13 @@ def simple_main_internal(createOptionParserFn, createCommandFn, mainOptions):
 
     # at this point we have whatever lock we require
     try:
-        simple_main_locked(parserOptions, parserArgs, createCommandFn, mainOptions)
+        simple_main_locked(parser, parserOptions, parserArgs, createCommandFn, mainOptions)
     finally:
         if sml is not None:
             sml.release()
 
 
-def simple_main_locked(parserOptions, parserArgs, createCommandFn, mainOptions):
+def simple_main_locked(parser, parserOptions, parserArgs, createCommandFn, mainOptions):
     """
     Not to be called externally -- use simple_main instead
     """
@@ -326,7 +326,6 @@ def simple_main_locked(parserOptions, parserArgs, createCommandFn, mainOptions):
     faultProberInterface.registerFaultProber(faultProberImplGpdb.GpFaultProberImplGpdb())
 
     commandObject = None
-    parser = None
 
     forceQuiet = mainOptions is not None and mainOptions.get("forceQuietOutput")
 

gpMgmt/bin/gppylib/operations/buildMirrorSegments.py

@@ -240,6 +240,9 @@ def __build_mirrors(self, actionName, gpEnv, gpArray):
 
         recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild)
 
+        # Remove any existing progress files for segments to be recovered
+        self.remove_existing_progress_files(recovery_info_by_host)
+
         self._run_setup_recovery(actionName, recovery_info_by_host)
 
         backout_map = self._update_config(recovery_info_by_host, gpArray)
@@ -292,13 +295,17 @@ def _update_config(self, recovery_info_by_host, gpArray):
             signal.signal(signal.SIGINT, old_handler)
             return backout_map
 
-    def _remove_progress_files(self, recovery_info_by_host, recovery_results):
+    # Remove any existing progress file of segments that will be recovered by
+    # current gprecoverseg execution. The files written by this run are left in
+    # place: gpstate reads them to report progress, and they are what an
+    # operator looks at after a recovery that went wrong.
+    def remove_existing_progress_files(self, recovery_info_by_host):
         remove_progress_file_cmds = []
         for hostName, recovery_info_list in recovery_info_by_host.items():
             for ri in recovery_info_list:
-                if recovery_results.was_bb_rewind_successful(ri.target_segment_dbid):
-                    remove_progress_file_cmds.append(self._get_remove_cmd(ri.progress_file, hostName))
-        self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=False)
+                remove_progress_file_cmds.append(self._get_remove_cmd("*dbid{}.out".format(ri.target_segment_dbid),
+                                                                      hostName))
+        self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=True)
 
     def _revert_config_update(self, recovery_results, backout_map):
         if len(backout_map) == 0:
@@ -485,7 +492,6 @@ def _run_recovery(self, action_name, recovery_info_by_host, gpEnv):
         recovery_results = RecoveryResult(action_name, completed_recovery_results, self.__logger)
         recovery_results.print_bb_rewind_and_start_errors()
 
-        self._remove_progress_files(recovery_info_by_host, recovery_results)
         return recovery_results
 
     def _do_recovery(self, recovery_info_by_host, gpEnv):

gpMgmt/bin/gppylib/programs/clsRecoverSegment.py

@@ -262,6 +262,15 @@ def run(self):
         if optionCnt > 1:
             raise ProgramArgumentValidationException("Only one of -i, -p, and -r may be specified")
 
+        # A full resynchronisation rebuilds the mirror from its primary, which
+        # is neither what -r (rebalance back to preferred roles) nor -p
+        # (recover onto a different host) asks for.
+        if self.__options.forceFullResynchronization:
+            if self.__options.rebalanceSegments:
+                raise ProgramArgumentValidationException("-F option is not supported with -r option")
+            if self.__options.newRecoverHosts is not None:
+                raise ProgramArgumentValidationException("-F option is not supported with -p option")
+
         faultProberInterface.getFaultProber().initializeProber(gpEnv.getCoordinatorPort())
 
         confProvider = configInterface.getConfigurationProvider().initializeProvider(gpEnv.getCoordinatorPort())
@@ -378,12 +387,16 @@ def signal_handler(sig, frame):
             if not mirrorBuilder.recover_mirrors(gpEnv, gpArray):
                 if self.termination_requested:
                     self.logger.error("gprecoverseg process was interrupted by the user.")
+                self.logger.error("gprecoverseg failed. Please check the output for more details.")
                 sys.exit(1)
 
             if self.termination_requested:
                 self.logger.info("Not able to terminate the recovery process since it has been completed successfully.")
 
             self.logger.info("********************************")
+            self.logger.info("Future gprecoverseg executions might remove the currently created pg_basebackup/pg_rewind/rsync "
+                             "progress files, please save these files if needed.")
+            self.logger.info("********************************")
             self.logger.info("Segments successfully recovered.")
             self.logger.info("********************************")