Re: [PATCH] fix DROP OPERATOR to reset links to itself on commutator and negator

diff --git a/src/backend/catalog/pg_operator.c b/src/backend/catalog/pg_operator.c
index 3cd1899..cd66823 100644
--- a/src/backend/catalog/pg_operator.c
+++ b/src/backend/catalog/pg_operator.c
@@ -54,8 +54,6 @@ static Oid OperatorShellMake(const char *operatorName,
 				  Oid leftTypeId,
 				  Oid rightTypeId);
 
-static void OperatorUpd(Oid baseId, Oid commId, Oid negId);
-
 static Oid get_other_operator(List *otherOp,
 				   Oid otherLeftTypeId, Oid otherRightTypeId,
 				   const char *operatorName, Oid operatorNamespace,
@@ -566,7 +564,7 @@ OperatorCreate(const char *operatorName,
 		commutatorId = operatorObjectId;
 
 	if (OidIsValid(commutatorId) || OidIsValid(negatorId))
-		OperatorUpd(operatorObjectId, commutatorId, negatorId);
+		OperatorUpd(operatorObjectId, commutatorId, negatorId, false);
 
 	return address;
 }
@@ -639,125 +637,158 @@ get_other_operator(List *otherOp, Oid otherLeftTypeId, Oid otherRightTypeId,
  * OperatorUpd
  *
  *	For a given operator, look up its negator and commutator operators.
- *	If they are defined, but their negator and commutator fields
- *	(respectively) are empty, then use the new operator for neg or comm.
- *	This solves a problem for users who need to insert two new operators
- *	which are the negator or commutator of each other.
+ *	When isDelete is false, update their negator and commutator operators to
+ *	point back to the given operator; when isDelete is true, update those
+ *	operators to no longer point back to the given operator.
+ *
+ *	The !isDelete case solves a problem for users who need to insert two new
+ *  operators which are the negator or commutator of each other, while the
+ *  isDelete case is important so as not to leave dangling OID links behind
+ *  after dropping an operator.
  */
-static void
-OperatorUpd(Oid baseId, Oid commId, Oid negId)
+void
+OperatorUpd(Oid baseId, Oid commId, Oid negId, bool isDelete)
 {
-	int			i;
 	Relation	pg_operator_desc;
-	HeapTuple	tup;
-	bool		nulls[Natts_pg_operator];
-	bool		replaces[Natts_pg_operator];
-	Datum		values[Natts_pg_operator];
-
-	for (i = 0; i < Natts_pg_operator; ++i)
-	{
-		values[i] = (Datum) 0;
-		replaces[i] = false;
-		nulls[i] = false;
-	}
+	HeapTuple	tup = NULL;
 
 	/*
-	 * check and update the commutator & negator, if necessary
-	 *
-	 * We need a CommandCounterIncrement here in case of a self-commutator
-	 * operator: we'll need to update the tuple that we just inserted.
+	 * If we're making an operator into its own commutator, then we need a
+	 * command-counter increment here, since we've just inserted the tuple
+	 * we're about to update.  But when we're dropping an operator, we can
+	 * skip this.
 	 */
-	CommandCounterIncrement();
+	if (!isDelete)
+		CommandCounterIncrement();
 
+	/* Open the relation. */
 	pg_operator_desc = heap_open(OperatorRelationId, RowExclusiveLock);
 
-	tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(commId));
+	/* Get a copy of the commutator's tuple. */
+	if (OidIsValid(commId))
+		tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(commId));
 
-	/*
-	 * if the commutator and negator are the same operator, do one update. XXX
-	 * this is probably useless code --- I doubt it ever makes sense for
-	 * commutator and negator to be the same thing...
-	 */
-	if (commId == negId)
+	/* Update the commutator's tuple if need be. */
+	if (HeapTupleIsValid(tup))
 	{
-		if (HeapTupleIsValid(tup))
+		Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);
+		bool		update_commutator = false;
+		bool		nulls[Natts_pg_operator];
+		bool		replaces[Natts_pg_operator];
+		Datum		values[Natts_pg_operator];
+
+		memset(values, 0, sizeof(values));
+		memset(replaces, 0, sizeof(replaces));
+		memset(nulls, 0, sizeof(nulls));
+
+		/*
+		 * Out of due caution, we only change the commutator's orpcom field
+		 * if it has the exact value we expected: InvalidOid when creating an
+		 * operator, and baseId when dropping one.
+		 */
+		if (isDelete && t->oprcom == baseId)
 		{
-			Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);
+			values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(InvalidOid);
+			replaces[Anum_pg_operator_oprcom - 1] = true;
+			update_commutator = true;
+		}
+		else if (!isDelete && !OidIsValid(t->oprcom))
+		{
+			values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(baseId);
+			replaces[Anum_pg_operator_oprcom - 1] = true;
+			update_commutator = true;
+		}
 
-			if (!OidIsValid(t->oprcom) || !OidIsValid(t->oprnegate))
+		/*
+		 * If the commutator is also the negator, which doesn't really make
+		 * sense but could perhaps happen, then both updates apply to the same
+		 * tuple and we should do just one update.  Handle that corner case
+		 * here.
+		 */
+		if (commId == negId)
+		{
+			if (isDelete && t->oprnegate == baseId)
+			{
+				values[Anum_pg_operator_oprnegate - 1] =
+					ObjectIdGetDatum(InvalidOid);
+				replaces[Anum_pg_operator_oprnegate - 1] = true;
+				update_commutator = true;
+			}
+			else if (!isDelete && !OidIsValid(t->oprnegate))
 			{
-				if (!OidIsValid(t->oprnegate))
-				{
-					values[Anum_pg_operator_oprnegate - 1] = ObjectIdGetDatum(baseId);
-					replaces[Anum_pg_operator_oprnegate - 1] = true;
-				}
-
-				if (!OidIsValid(t->oprcom))
-				{
-					values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(baseId);
-					replaces[Anum_pg_operator_oprcom - 1] = true;
-				}
-
-				tup = heap_modify_tuple(tup,
-										RelationGetDescr(pg_operator_desc),
-										values,
-										nulls,
-										replaces);
-
-				simple_heap_update(pg_operator_desc, &tup->t_self, tup);
-
-				CatalogUpdateIndexes(pg_operator_desc, tup);
+				values[Anum_pg_operator_oprnegate - 1] =
+					ObjectIdGetDatum(baseId);
+				replaces[Anum_pg_operator_oprnegate - 1] = true;
+				update_commutator = true;
 			}
 		}
 
-		heap_close(pg_operator_desc, RowExclusiveLock);
-
-		return;
-	}
-
-	/* if commutator and negator are different, do two updates */
-
-	if (HeapTupleIsValid(tup) &&
-		!(OidIsValid(((Form_pg_operator) GETSTRUCT(tup))->oprcom)))
-	{
-		values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(baseId);
-		replaces[Anum_pg_operator_oprcom - 1] = true;
-
-		tup = heap_modify_tuple(tup,
-								RelationGetDescr(pg_operator_desc),
-								values,
-								nulls,
-								replaces);
-
-		simple_heap_update(pg_operator_desc, &tup->t_self, tup);
-
-		CatalogUpdateIndexes(pg_operator_desc, tup);
-
-		values[Anum_pg_operator_oprcom - 1] = (Datum) NULL;
-		replaces[Anum_pg_operator_oprcom - 1] = false;
+		/* If any columns were found to need modification, update tuple. */
+		if (update_commutator)
+		{
+			tup = heap_modify_tuple(tup,
+									RelationGetDescr(pg_operator_desc),
+									values,
+									nulls,
+									replaces);
+			simple_heap_update(pg_operator_desc, &tup->t_self, tup);
+			CatalogUpdateIndexes(pg_operator_desc, tup);
+		}
 	}
 
-	/* check and update the negator, if necessary */
-
-	tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(negId));
-
-	if (HeapTupleIsValid(tup) &&
-		!(OidIsValid(((Form_pg_operator) GETSTRUCT(tup))->oprnegate)))
+	/*
+	 * Assuming (as will almost always be the case) that the commutator and
+	 * negator are different, we now need to update the negator's tuple,
+	 * assuming there is a negator.
+	 */
+	tup = NULL;
+	if (OidIsValid(negId) && commId != negId)
+		tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(negId));
+	if (HeapTupleIsValid(tup))
 	{
-		values[Anum_pg_operator_oprnegate - 1] = ObjectIdGetDatum(baseId);
-		replaces[Anum_pg_operator_oprnegate - 1] = true;
+		Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);
+		bool		update_negator = false;
+		bool		nulls[Natts_pg_operator];
+		bool		replaces[Natts_pg_operator];
+		Datum		values[Natts_pg_operator];
 
-		tup = heap_modify_tuple(tup,
-								RelationGetDescr(pg_operator_desc),
-								values,
-								nulls,
-								replaces);
+		memset(values, 0, sizeof(values));
+		memset(replaces, 0, sizeof(replaces));
+		memset(nulls, 0, sizeof(nulls));
 
-		simple_heap_update(pg_operator_desc, &tup->t_self, tup);
+		/*
+		 * Out of due caution, we only change the negators's orpneg field
+		 * if it has the exact value we expected: InvalidOid when creating an
+		 * operator, and baseId when dropping one.
+		 */
+		if (isDelete && t->oprnegate == baseId)
+		{
+			values[Anum_pg_operator_oprnegate - 1] =
+				ObjectIdGetDatum(InvalidOid);
+			replaces[Anum_pg_operator_oprnegate - 1] = true;
+			update_negator = true;
+		}
+		else if (!isDelete && !OidIsValid(t->oprnegate))
+		{
+			values[Anum_pg_operator_oprnegate - 1] = ObjectIdGetDatum(baseId);
+			replaces[Anum_pg_operator_oprnegate - 1] = true;
+			update_negator = true;
+		}
 
-		CatalogUpdateIndexes(pg_operator_desc, tup);
+		/* If any columns were found to need modification, update tuple. */
+		if (update_negator)
+		{
+			tup = heap_modify_tuple(tup,
+									RelationGetDescr(pg_operator_desc),
+									values,
+									nulls,
+									replaces);
+			simple_heap_update(pg_operator_desc, &tup->t_self, tup);
+			CatalogUpdateIndexes(pg_operator_desc, tup);
+		}
 	}
 
+	/* Close relation and release catalog lock */
 	heap_close(pg_operator_desc, RowExclusiveLock);
 }
 
diff --git a/src/backend/commands/operatorcmds.c b/src/backend/commands/operatorcmds.c
index 664e5d7..af95b9a 100644
--- a/src/backend/commands/operatorcmds.c
+++ b/src/backend/commands/operatorcmds.c
@@ -339,15 +339,32 @@ ValidateJoinEstimator(List *joinName)
 void
 RemoveOperatorById(Oid operOid)
 {
-	Relation	relation;
-	HeapTuple	tup;
+	Relation			relation;
+	HeapTuple			tup;
+	Form_pg_operator	op;
 
 	relation = heap_open(OperatorRelationId, RowExclusiveLock);
 
 	tup = SearchSysCache1(OPEROID, ObjectIdGetDatum(operOid));
+	op = (Form_pg_operator) GETSTRUCT(tup);
 	if (!HeapTupleIsValid(tup)) /* should not happen */
 		elog(ERROR, "cache lookup failed for operator %u", operOid);
 
+	/*
+	 * Reset links on commutator and negator. Only do that if either
+	 * oprcom or oprnegate is set and given operator is not self-commutator.
+	 * For self-commutator with negator, prevent meaningful updates of the
+	 * same tuple by sending InvalidOid.
+	 * Since operator can't be its own negator, we don't need this check for
+	 * negator: if there is oprnegate, it should be updated.
+	 */
+	if (OidIsValid(op->oprnegate) ||
+		(OidIsValid(op->oprcom) && operOid != op->oprcom))
+		OperatorUpd(operOid,
+					operOid == op->oprcom ? InvalidOid : op->oprcom,
+					op->oprnegate,
+					true);
+
 	simple_heap_delete(relation, &tup->t_self);
 
 	ReleaseSysCache(tup);
diff --git a/src/include/catalog/pg_operator_fn.h b/src/include/catalog/pg_operator_fn.h
index 5315e19..eb8d011 100644
--- a/src/include/catalog/pg_operator_fn.h
+++ b/src/include/catalog/pg_operator_fn.h
@@ -30,5 +30,6 @@ extern ObjectAddress OperatorCreate(const char *operatorName,
 			   bool canHash);
 
 extern ObjectAddress makeOperatorDependencies(HeapTuple tuple, bool isUpdate);
+extern void OperatorUpd(Oid baseId, Oid commId, Oid negId, bool isDelete);
 
 #endif   /* PG_OPERATOR_FN_H */
diff --git a/src/test/regress/expected/drop_operator.out b/src/test/regress/expected/drop_operator.out
new file mode 100644
index 0000000..cc8f5e7
--- /dev/null
+++ b/src/test/regress/expected/drop_operator.out
@@ -0,0 +1,61 @@
+CREATE OPERATOR === (
+        PROCEDURE = int8eq,
+        LEFTARG = bigint,
+        RIGHTARG = bigint,
+        COMMUTATOR = ===
+);
+CREATE OPERATOR !== (
+        PROCEDURE = int8ne,
+        LEFTARG = bigint,
+        RIGHTARG = bigint,
+        NEGATOR = ===,
+        COMMUTATOR = !==
+);
+DROP OPERATOR !==(bigint, bigint);
+SELECT  ctid, oprcom
+FROM    pg_catalog.pg_operator fk
+WHERE   oprcom != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprcom);
+ ctid | oprcom 
+------+--------
+(0 rows)
+
+SELECT  ctid, oprnegate
+FROM    pg_catalog.pg_operator fk
+WHERE   oprnegate != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprnegate);
+ ctid | oprnegate 
+------+-----------
+(0 rows)
+
+DROP OPERATOR ===(bigint, bigint);
+CREATE OPERATOR <| (
+        PROCEDURE = int8lt,
+        LEFTARG = bigint,
+        RIGHTARG = bigint
+);
+CREATE OPERATOR |> (
+        PROCEDURE = int8gt,
+        LEFTARG = bigint,
+        RIGHTARG = bigint,
+        NEGATOR = <|,
+        COMMUTATOR = <|
+);
+DROP OPERATOR |>(bigint, bigint);
+SELECT  ctid, oprcom
+FROM    pg_catalog.pg_operator fk
+WHERE   oprcom != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprcom);
+ ctid | oprcom 
+------+--------
+(0 rows)
+
+SELECT  ctid, oprnegate
+FROM    pg_catalog.pg_operator fk
+WHERE   oprnegate != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprnegate);
+ ctid | oprnegate 
+------+-----------
+(0 rows)
+
+DROP OPERATOR <|(bigint, bigint);
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index bec0316..731677f 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -79,7 +79,7 @@ ignore: random
 # ----------
 # Another group of parallel tests
 # ----------
-test: select_into select_distinct select_distinct_on select_implicit select_having subselect union case join aggregates transactions random portals arrays btree_index hash_index update namespace prepared_xacts delete
+test: select_into select_distinct select_distinct_on select_implicit select_having subselect union case join aggregates transactions random portals arrays btree_index hash_index update namespace prepared_xacts delete drop_operator
 
 # ----------
 # Another group of parallel tests
diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule
index 7e9b319..4e2fe83 100644
--- a/src/test/regress/serial_schedule
+++ b/src/test/regress/serial_schedule
@@ -115,6 +115,7 @@ test: object_address
 test: tablesample
 test: alter_generic
 test: alter_operator
+test: drop_operator
 test: misc
 test: psql
 test: async
diff --git a/src/test/regress/sql/drop_operator.sql b/src/test/regress/sql/drop_operator.sql
new file mode 100644
index 0000000..cc62cfa
--- /dev/null
+++ b/src/test/regress/sql/drop_operator.sql
@@ -0,0 +1,56 @@
+CREATE OPERATOR === (
+        PROCEDURE = int8eq,
+        LEFTARG = bigint,
+        RIGHTARG = bigint,
+        COMMUTATOR = ===
+);
+
+CREATE OPERATOR !== (
+        PROCEDURE = int8ne,
+        LEFTARG = bigint,
+        RIGHTARG = bigint,
+        NEGATOR = ===,
+        COMMUTATOR = !==
+);
+
+DROP OPERATOR !==(bigint, bigint);
+
+SELECT  ctid, oprcom
+FROM    pg_catalog.pg_operator fk
+WHERE   oprcom != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprcom);
+
+SELECT  ctid, oprnegate
+FROM    pg_catalog.pg_operator fk
+WHERE   oprnegate != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprnegate);
+
+DROP OPERATOR ===(bigint, bigint);
+
+CREATE OPERATOR <| (
+        PROCEDURE = int8lt,
+        LEFTARG = bigint,
+        RIGHTARG = bigint
+);
+
+CREATE OPERATOR |> (
+        PROCEDURE = int8gt,
+        LEFTARG = bigint,
+        RIGHTARG = bigint,
+        NEGATOR = <|,
+        COMMUTATOR = <|
+);
+
+DROP OPERATOR |>(bigint, bigint);
+
+SELECT  ctid, oprcom
+FROM    pg_catalog.pg_operator fk
+WHERE   oprcom != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprcom);
+
+SELECT  ctid, oprnegate
+FROM    pg_catalog.pg_operator fk
+WHERE   oprnegate != 0 AND
+        NOT EXISTS(SELECT 1 FROM pg_catalog.pg_operator pk WHERE pk.oid = fk.oprnegate);
+
+DROP OPERATOR <|(bigint, bigint);