diff --git a/gpMgmt/bin/gpcheckcat b/gpMgmt/bin/gpcheckcat
index 00c2e4b21f3..5c0036d96a0 100755
--- a/gpMgmt/bin/gpcheckcat
+++ b/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,207 @@ def checkOrphanedToastTables():
                                issue_type="orphaned_toast_tables",
                                description='Repairing orphaned TOAST tables')
 
+# The mix_distribution_policy check below -- fetch_guc_value(),
+# generateDistPolicyQueryFile(), checkMixDistPolicy() and their helpers --
+# is cherry-picked from Greenplum Database (greenplum-db/gpdb-archive @ 482967c1b4),
+# gpMgmt/bin/gpcheckcat.
+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 +2286,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 +2579,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']
diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py
index ccdfb03a7ad..32fe6cbbffc 100755
--- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py
+++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py
@@ -411,6 +411,178 @@ def count_joins():
                 self.num_batches += 1
                 self.num_joins = 0
                 self.num_starts = 0
+
+    # gpcheckcat talks to the database through PyGreSQL: connect2() returns a
+    # pg connection, conn.query() runs the statement and getresult() fetches
+    # the rows.  Greenplum's versions of these tests mock a psycopg2 cursor,
+    # which this gpcheckcat never opens.
+    @patch('gpcheckcat.connect2')
+    def test_checkMixDistPolicy_with_error_on_execution(self, mock_connect2):
+        mock_connect2.return_value.query.side_effect = Exception("Simulated error during execution")
+
+        self.subject.checkMixDistPolicy()
+
+        self.assertEqual(mock_connect2.return_value.query.call_count, 1)
+        self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR)
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.connect2')
+    def test_checkMixDistPolicy_exception_on_connect(self, mock_connect2):
+        mock_connect2.side_effect = Exception("Simulated error during connection")
+
+        self.subject.checkMixDistPolicy()
+
+        self.assertEqual(mock_connect2.call_count, 1)
+        self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR)
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.connect2')
+    def test_checkMixDistPolicy_exception_on_getresult(self, mock_connect2):
+        mock_connect2.return_value.query.return_value.getresult.side_effect = Exception("Simulated error fetching rows")
+
+        self.subject.checkMixDistPolicy()
+
+        self.assertEqual(mock_connect2.return_value.query.call_count, 1)
+        self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR)
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.generateDistPolicyQueryFile')
+    @patch('gpcheckcat.connect2')
+    def test_checkMixDistPolicy_reports_mixed_policy(self, mock_connect2, mock_generate):
+        # 1 legacy distribution opclass out of 3 in use
+        mock_connect2.return_value.query.return_value.getresult.return_value = [(1, 3)]
+
+        self.subject.checkMixDistPolicy()
+
+        mock_generate.assert_called_once_with()
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.fetch_guc_value', return_value='off')
+    @patch('gpcheckcat.connect2')
+    def test_checkMixDistPolicy_passes_without_legacy_policy(self, mock_connect2, mock_fetch):
+        mock_connect2.return_value.query.return_value.getresult.return_value = [(0, 3)]
+
+        self.subject.checkMixDistPolicy()
+
+        mock_fetch.assert_called_once_with('gp_use_legacy_hashops')
+        self.assertTrue(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.connect2')
+    def test_fetch_guc_value_returns_value(self, mock_connect2):
+        mock_connect2.return_value.query.return_value.getresult.return_value = [('on',)]
+
+        self.assertEqual(self.subject.fetch_guc_value("gp_use_legacy_hashops"), 'on')
+        self.assertIn('show gp_use_legacy_hashops', mock_connect2.return_value.query.call_args[0][0])
+        self.assertTrue(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.connect2')
+    def test_fetch_guc_value_with_error_on_execution(self, mock_connect2):
+        mock_connect2.return_value.query.side_effect = Exception("Simulated error during execution")
+
+        self.assertIsNone(self.subject.fetch_guc_value("gp_use_legacy_hashops"))
+
+        self.assertEqual(mock_connect2.return_value.query.call_count, 1)
+        self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR)
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.connect2')
+    def test_fetch_guc_value_exception_on_connect(self, mock_connect2):
+        mock_connect2.side_effect = Exception("Simulated error during connection")
+
+        self.assertIsNone(self.subject.fetch_guc_value("gp_use_legacy_hashops"))
+
+        self.assertEqual(mock_connect2.call_count, 1)
+        self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR)
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    @patch('gpcheckcat.connect2')
+    def test_fetch_guc_value_exception_on_getresult(self, mock_connect2):
+        mock_connect2.return_value.query.return_value.getresult.side_effect = Exception("Simulated error fetching rows")
+
+        self.assertIsNone(self.subject.fetch_guc_value("gp_use_legacy_hashops"))
+
+        self.assertEqual(mock_connect2.return_value.query.call_count, 1)
+        self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR)
+        self.assertFalse(self.subject.GV.checkStatus)
+
+    def test_generateDistPolicyQueryFile(self):
+        # Call the function
+        self.subject.generateDistPolicyQueryFile()
+
+        # Check if the file is created
+        self.assertTrue(os.path.exists('gpcheckcat.distpolicy.sql'))
+
+        # Read the file and check if it contains the expected query
+        with open('gpcheckcat.distpolicy.sql', 'r') as fp:
+            generated_query = fp.read()
+
+        expected_query = '''
+     -- 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);
+           '''
+        self.assertEqual.__self__.maxDiff = None
+        self.assertTrue(generated_query.strip() == expected_query.strip())
 class Global():
     def __init__(self):
         self.opt = {}
diff --git a/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_legacy_hash_ops_tables.sql b/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_legacy_hash_ops_tables.sql
new file mode 100644
index 00000000000..4cd9a745a60
--- /dev/null
+++ b/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_legacy_hash_ops_tables.sql
@@ -0,0 +1,37 @@
+-- Cherry-picked from Greenplum Database, unmodified.
+-- Source: greenplum-db/gpdb-archive @ 482967c1b4
+--         gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_legacy_hash_ops_tables.sql
+
+set gp_use_legacy_hashops = 1;
+
+create table t_old(a int, b int, c int) distributed by (a, b);
+create table t1_old(a int, b int, c int) distributed by (a, b);
+create table t_replicate_old(a int , b int) distributed replicated;
+create table t_random_old(a int , b int) distributed randomly;
+
+CREATE TABLE rank_old (id int, rank int, year int, gender
+        char(1), count int)
+DISTRIBUTED BY (id)
+PARTITION BY RANGE (year)
+( START (2006) END (2016) EVERY (1),
+          DEFAULT PARTITION extra );
+
+
+CREATE OR REPLACE FUNCTION random_between(low INT ,high INT)
+   RETURNS INT AS
+$$
+BEGIN
+           RETURN floor(random()* (high-low + 1) + low);
+END;
+$$ language 'plpgsql' STRICT;
+
+insert into rank_old
+select i, i, random_between(2005, 2017), 'g', i
+from generate_series(1, 100000)i;
+
+-- some special characters in column names
+create table t_space("a col" int);
+create table t_dot("a.col" int);
+create table t_dash("a-col" int);
+create table t_multispecial("a col" int, "b.col" int, "c-col" int) distributed by ("a col", "b.col", "c-col");
+
diff --git a/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_non_legacy_hashops_tables.sql b/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_non_legacy_hashops_tables.sql
new file mode 100644
index 00000000000..bdd5a23a75e
--- /dev/null
+++ b/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_non_legacy_hashops_tables.sql
@@ -0,0 +1,30 @@
+-- Cherry-picked from Greenplum Database, unmodified.
+-- Source: greenplum-db/gpdb-archive @ 482967c1b4
+--         gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_non_legacy_hashops_tables.sql
+
+set gp_use_legacy_hashops = 0;
+
+create table t_new(a int, b int, c int) distributed by (a, b);
+create table t1_new(a int, b int, c int) distributed by (a, b);
+create table t_replicate_new(a int , b int) distributed replicated;
+create table t_random_new(a int , b int) distributed randomly;
+
+CREATE TABLE rank_new (id int, rank int, year int, gender
+        char(1), count int)
+DISTRIBUTED BY (id)
+PARTITION BY RANGE (year)
+( START (2006) END (2016) EVERY (1),
+          DEFAULT PARTITION extra );
+
+
+CREATE OR REPLACE FUNCTION random_between(low INT ,high INT)
+   RETURNS INT AS
+$$
+BEGIN
+           RETURN floor(random()* (high-low + 1) + low);
+END;
+$$ language 'plpgsql' STRICT;
+
+insert into rank_new
+select i, i, random_between(2005, 2017), 'g', i
+from generate_series(1, 100000)i;