[BABEL] Fix: Rectified removal and updates of incorrect tuples from ENR catalogs (babelfish-for-postgresql#700)

PG18 introduced a new constraint which is created for all non-null
attributes including p_keys. So now, an attribute which is a p_key will
create two constraints - p_key and not-null constraint.

When adding these constraints on the catalog in ENR, the condition tf2.oid > tf1.oid
will keep the catalog list sorted, however, while removing or updating
any tuple in this catalog list, we compare oids in a way which is
interpreted as remove/update oid of tuple which has an oid greater than
or equal to the requested oid. This leads to catalog corruption since we
are removing/updating a different tuple than the requested one.

This change rectifies this condition and breaks it down based on the
operation being performed.

Task: BABEL-6211

Signed-off-by: Ayush Shah <ayushdsh@amazon.com>

Author: ayushdsh

src/backend/utils/misc/queryenvironment.c

@@ -788,6 +788,53 @@ bool ENRUpdateTuple(Relation rel, HeapTuple tup)
 	return _ENR_tuple_operation(rel, tup, ENR_OP_UPDATE, false, false);
 }
 
+/*
+ * Find matching tuple in ENR catalog list for DROP/UPDATE operations.
+ * Returns the ListCell containing the matching tuple, or NULL if not found.
+ */
+static ListCell *
+find_tuple_in_enr_catalog(List *cattups, HeapTuple search_tup, Oid catalog_oid)
+{
+	ListCell *lc;
+
+	foreach(lc, cattups)
+	{
+		HeapTuple enr_tup = (HeapTuple) lfirst(lc);
+		
+		switch (catalog_oid)
+		{
+			case IndexRelationId:
+			{
+				Form_pg_index idx1 = (Form_pg_index) GETSTRUCT(search_tup);
+				Form_pg_index idx2 = (Form_pg_index) GETSTRUCT(enr_tup);
+				if (idx1->indexrelid == idx2->indexrelid)
+					return lc;
+				break;
+			}
+			case AttrDefaultRelationId:
+			{
+				Form_pg_attrdef def1 = (Form_pg_attrdef) GETSTRUCT(search_tup);
+				Form_pg_attrdef def2 = (Form_pg_attrdef) GETSTRUCT(enr_tup);
+				if (def1->oid == def2->oid)
+					return lc;
+				break;
+			}
+			case ConstraintRelationId:
+			{
+				Form_pg_constraint con1 = (Form_pg_constraint) GETSTRUCT(search_tup);
+				Form_pg_constraint con2 = (Form_pg_constraint) GETSTRUCT(enr_tup);
+				if (con1->oid == con2->oid)
+					return lc;
+				break;
+			}
+			default:
+				elog(ERROR, "Unsupported catalog OID %u for tuple matching", catalog_oid);
+		}
+	}
+	
+	return NULL;
+}
+
 /*
  * Workhorse for add/update/drop tuples in the ENR.
  *
@@ -914,8 +961,10 @@ static bool _ENR_tuple_operation(Relation catalog_rel, HeapTuple tup, ENRTupleOp
 				rel_oid = ((Form_pg_index) GETSTRUCT(tup))->indrelid;
 				if ((enr = get_ENR_withoid(queryEnv, rel_oid, ENR_TSQL_TEMP, false))) {
 					list_ptr = &enr->md.cattups[ENR_CATTUP_INDEX];
-					lc = list_head(enr->md.cattups[ENR_CATTUP_INDEX]);
 					ret = true;
+					
+					if (op == ENR_OP_DROP || op == ENR_OP_UPDATE)
+						lc = find_tuple_in_enr_catalog(*list_ptr, tup, catalog_oid);
 
 					if ((op == ENR_OP_ADD || op == ENR_OP_UPDATE) && HeapTupleIsValid(tup))
 					{
@@ -994,20 +1043,19 @@ static bool _ENR_tuple_operation(Relation catalog_rel, HeapTuple tup, ENRTupleOp
 			case ConstraintRelationId:
 				rel_oid = ((Form_pg_constraint) GETSTRUCT(tup))->conrelid;
 				if ((enr = get_ENR_withoid(queryEnv, rel_oid, ENR_TSQL_TEMP, false))) {
-					Form_pg_constraint tf1, tf2; /* tuple forms*/
-					ListCell *curlc;
-
 					list_ptr = &enr->md.cattups[ENR_CATTUP_CONSTRAINT];
-					tf1 = (Form_pg_constraint) GETSTRUCT(tup);
-					foreach(curlc, enr->md.cattups[ENR_CATTUP_CONSTRAINT]) {
-						tf2 = (Form_pg_constraint) GETSTRUCT((HeapTuple) lfirst(curlc));
-						if (tf2->oid >= tf1->oid) {
-							lc = curlc;
-							insert_at = foreach_current_index(curlc) + 1;
+					ret = true;
+					
+					switch (op)
+					{
+						case ENR_OP_ADD:
+							insert_at = list_length(enr->md.cattups[ENR_CATTUP_CONSTRAINT]);
+							break;
+						case ENR_OP_DROP:
+						case ENR_OP_UPDATE:
+							lc = find_tuple_in_enr_catalog(*list_ptr, tup, catalog_oid);
 							break;
-						}
 					}
-					ret = true;
 				}
 				break;
 			case StatisticRelationId:
@@ -1060,8 +1108,10 @@ static bool _ENR_tuple_operation(Relation catalog_rel, HeapTuple tup, ENRTupleOp
 				rel_oid = ((Form_pg_attrdef) GETSTRUCT(tup))->adrelid;
 				if ((enr = get_ENR_withoid(queryEnv, rel_oid, ENR_TSQL_TEMP, false))) {
 					list_ptr = &enr->md.cattups[ENR_CATTUP_ATTR_DEF_REL];
-					lc = list_head(enr->md.cattups[ENR_CATTUP_ATTR_DEF_REL]);
 					ret = true;
+					
+					if (op == ENR_OP_DROP || op == ENR_OP_UPDATE)
+						lc = find_tuple_in_enr_catalog(*list_ptr, tup, catalog_oid);
 				}
 				break;
 			default:
@@ -1850,4 +1900,4 @@ bool has_existing_enr_relations()
 	}
 
 	return false;
-}
+}