This is an automated email from the ASF dual-hosted git repository. tuhaihe pushed a commit to branch add-behave-test-ci in repository https://gitbox.apache.org/repos/asf/cloudberry.git
commit 70abfd57849bc12d192c4e543bd8c311f7fac272 Author: Dianjin Wang <[email protected]> AuthorDate: Fri Sep 11 12:23:52 2026 +0800 gpcheckcat: add the mix_distribution_policy and AO lastrownums checks Back-port two checks from Greenplum that Cloudberry was missing. mix_distribution_policy reports tables whose distribution policy mixes legacy and non-legacy hash opclasses, and cross-checks the result against the gp_use_legacy_hashops GUC so the operator is told which of the two states is inconsistent. checkAOLastrownums reports pg_fastsequence.last_sequence values that disagree with gp_fastsequence for AO and AOCO tables. Add the SQL fixtures the corresponding Behave scenarios load. --- gpMgmt/bin/gpcheckcat | 297 ++++++++++++++++++++- .../bin/gppylib/test/unit/test_unit_gpcheckcat.py | 170 ++++++++++++ .../gpcheckcat/create_legacy_hash_ops_tables.sql | 33 +++ .../create_non_legacy_hashops_tables.sql | 26 ++ 4 files changed, 525 insertions(+), 1 deletion(-) diff --git a/gpMgmt/bin/gpcheckcat b/gpMgmt/bin/gpcheckcat index 00c2e4b21f3..a4c1ede8bfa 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) @@ -1525,6 +1528,84 @@ def checkAOSegVpinfo(): # ------------------------------------------------------------------------------- +class checkAOLastrownumThread(execThread): + def __init__(self, cfg, db): + execThread.__init__(self, cfg, db, None) + + # pg_attribute_encoding.lastrownums[segno], if exists, should have a corresponding entry in + # gp_fastsequence with an objid same as segno. And the value of + # pg_attribute_encoding.lastrownums[segno] should fall in the range of [0, {last_sequence}] + # where {last_sequence} is the current gp_fastsequence value with the corresponding objid. + # Note that objmod starts from 0 but the array index starts from 1. + def run(self): + aolastrownum_query = """ + SELECT + c.relname, + ao.relid, + ae.attnum, + ae.lastrownums, + f.objmod, + f.last_sequence, + ae.lastrownums[f.objmod + 1] AS lastrownum + FROM + pg_attribute_encoding ae + JOIN pg_appendonly ao ON ae.attrelid = ao.relid + LEFT JOIN gp_fastsequence f ON ao.segrelid = f.objid + JOIN pg_class c ON ao.relid = c.oid + WHERE + f.last_sequence IS NULL + OR f.last_sequence < ae.lastrownums[f.objmod + 1] + OR ae.lastrownums[f.objmod + 1] < 0; + """ + + try: + curs = self.db.query(aolastrownum_query) + rows = curs.getresult() + + if len(rows) == 0: + logger.info('[OK] AO lastrownums check for pg_attribute_encoding') + else: + GV.checkStatus = False + # we could not fix this issue automatically + setError(ERROR_NOREPAIR) + logger.info('[FAIL] AO lastrownums check for pg_attribute_encoding') + for relname, relid, attnum, lastrownums, objmod, last_sequence, last_rownum in rows: + logger.error(" found inconsistent last_rownum {rownum} with last_sequence {seqnum} of aoseg {segno} for table '{relname}' attribute {attnum} on segment {content}" + .format(rownum = last_rownum, + seqnum = last_sequence, + segno = objmod, + relname = relname, + attnum = attnum, + content = self.cfg['content'])) + + except Exception as e: + GV.checkStatus = False + self.error = e + +# for test "ao_lastrownums" +def checkAOLastrownums(): + threads = [] + i = 1 + # parallelise check + for dbid in GV.cfg: + cfg = GV.cfg[dbid] + conn = connect2(cfg) + thread = checkAOLastrownumThread(cfg, conn) + thread.start() + logger.debug('launching check thread %s for dbid %i' % + (thread.name, dbid)) + threads.append(thread) + + if (i % GV.opt['-B']) == 0: + processThread(threads) + threads = [] + + i += 1 + + processThread(threads) + +# ------------------------------------------------------------------------------- + # Exclude these tuples from the catalog table scan # pg_depend: classid 2603 (pg_amproc) is excluded because their OIDs can be # nonsynchronous (catalog.c:RelationNeedsSynchronizedOIDs()) @@ -1948,6 +2029,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 +2360,24 @@ all_checks = { "version": 'main', "order": 15, "online": False + }, + "ao_lastrownums": + { + "description": "Check that lastrownums in pg_attribute_encoding is consistent with gp_fastsequence", + "fn": lambda: checkAOLastrownums(), + "version": 'main', + "order": 16, + "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 } + } diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py index ccdfb03a7ad..e4ecdf616a0 100755 --- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py +++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpcheckcat.py @@ -411,6 +411,176 @@ class GpCheckCatTestCase(GpTestCase): self.num_batches += 1 self.num_joins = 0 self.num_starts = 0 + + @patch('gpcheckcat.connect2') + def test_checkMixDistPolicy_with_error_on_execution(self,mock_connect2): + # Mocking the database connection to raise an exception during execution + + mock_cursor = Mock() + mock_connect2.return_value.cursor.return_value.__enter__.return_value = mock_cursor + mock_cursor.execute.side_effect = Exception("Simulated error during execution") + + # Call the function to test + self.subject.checkMixDistPolicy() + + # Assertions + self.assertEqual(mock_cursor.execute.call_count, 1) + self.assertFalse(self.subject.GV.checkStatus) + + @patch('gpcheckcat.connect2') + def test_checkMixDistPolicy_exception_on_connect(self, mock_connect2): + # Mocking the database connection to raise an exception during connection + mock_connect2.side_effect = Exception("Simulated error during connection") + + # Call the function to test + self.subject.checkMixDistPolicy() + + # Assertions + self.assertEqual(mock_connect2.call_count, 1) + self.assertFalse(self.subject.GV.checkStatus) + + @patch('gpcheckcat.connect2') + def test_checkMixDistPolicy_exception_on_cursor_enter(self, mock_connect2): + # Mocking the database connection to raise an exception when entering the cursor context + mock_connect2.return_value.cursor.return_value.__enter__.side_effect = Exception("Simulated error entering cursor context") + + # Call the function to test + self.subject.checkMixDistPolicy() + + # Assertions + self.assertEqual(mock_connect2.call_count, 1) + self.assertFalse(self.subject.GV.checkStatus) + + @patch('gpcheckcat.connect2') + def test_fetch_guc_value_with_error_on_execution(self,mock_connect2): + # Mocking the database connection to raise an exception during execution + + mock_cursor = Mock() + mock_connect2.return_value.cursor.return_value.__enter__.return_value = mock_cursor + mock_cursor.execute.side_effect = Exception("Simulated error during execution") + + # Call the function to test + self.subject.fetch_guc_value("gp_use_legacy_hashops") + + # Assertions + self.assertEqual(mock_cursor.execute.call_count, 1) + self.assertEqual(mock_cursor.fetchone.call_count, 0) + 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): + # Mocking the database connection to raise an exception during connection + mock_cursor = Mock() + mock_connect2.return_value.cursor.return_value.__enter__.return_value = mock_cursor + mock_connect2.side_effect = Exception("Simulated error during execution") + + # Call the function to test + self.subject.fetch_guc_value("gp_use_legacy_hashops") + + # Assertions + self.assertEqual(mock_cursor.execute.call_count, 0) + self.assertEqual(mock_connect2.call_count, 1) + self.assertEqual(mock_cursor.fetchone.call_count, 0) + 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_cursor_enter(self, mock_connect2): + # Mocking the database connection to raise an exception on cursor context + mock_cursor = Mock() + #mock_connect2.return_value.cursor.return_value.__enter__.return_value = mock_cursor + mock_connect2.return_value.cursor.return_value = mock_cursor + mock_cursor.side_effect = Exception("Simulated error during execution") + + # Call the function to test + self.subject.fetch_guc_value("gp_use_legacy_hashops") + + # Assertions + self.assertEqual(mock_connect2.call_count, 1) + self.assertEqual(mock_cursor.execute.call_count, 0) + self.assertFalse(self.subject.GV.checkStatus) + self.assertEqual(mock_cursor.fetchone.call_count, 0) + self.subject.setError.assert_any_call(self.subject.ERROR_NOREPAIR) + + 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..d79ac75a184 --- /dev/null +++ b/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_legacy_hash_ops_tables.sql @@ -0,0 +1,33 @@ +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..cf6e89d7d9e --- /dev/null +++ b/gpMgmt/test/behave/mgmt_utils/steps/data/gpcheckcat/create_non_legacy_hashops_tables.sql @@ -0,0 +1,26 @@ +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; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
