Fix DROP OPERATOR to reset oprcom/oprnegate links to the dropped operator.

This avoids leaving dangling links in pg_operator; which while fairly
harmless are also unsightly.

While we're at it, simplify OperatorUpd, which went through
heap_modify_tuple for no very good reason considering it had already made
a tuple copy it could just scribble on.

Roma Sokolov, reviewed by Tomas Vondra, additional hacking by Robert Haas
and myself.

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,125 @@ 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
+ *	When isDelete is false, update their negator and commutator fields to
- *	(respectively) are empty, then use the new operator for neg or comm.
+ *	point back to the given operator; when isDelete is true, update those
- *	This solves a problem for users who need to insert two new operators
+ *	fields to no longer point back to the given operator.
- *	which are the negator or commutator of each other.
+ *
+ *	The !isDelete case solves a problem for users who need to insert two new
+ *	operators that are the negator or commutator of each other, while the
+ *	isDelete case is needed so as not to leave dangling OID links behind
+ *	after dropping an operator.
  */
-static void
+void
-OperatorUpd(Oid baseId, Oid commId, Oid negId)
+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;
-	}
 
 	/*
-	 * check and update the commutator & negator, if necessary
+	 * 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 need a CommandCounterIncrement here in case of a self-commutator
+	 * we're about to update.  But when we're dropping an operator, we can
-	 * operator: we'll need to update the tuple that we just inserted.
+	 * skip this because we're at the beginning of the command.
 	 */
-	CommandCounterIncrement();
+	if (!isDelete)
+		CommandCounterIncrement();
 
+	/* Open the relation. */
 	pg_operator_desc = heap_open(OperatorRelationId, RowExclusiveLock);
 
-	tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(commId));
+	/* Get a writable copy of the commutator's tuple. */
+	if (OidIsValid(commId))
+		tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(commId));
+	else
+		tup = NULL;
 
-	/*
+	/* Update the commutator's tuple if need be. */
-	 * if the commutator and negator are the same operator, do one update. XXX
+	if (HeapTupleIsValid(tup))
-	 * this is probably useless code --- I doubt it ever makes sense for
-	 * commutator and negator to be the same thing...
-	 */
-	if (commId == negId)
 	{
-		if (HeapTupleIsValid(tup))
+		Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);
+		bool		update_commutator = false;
+
+		/*
+		 * Out of due caution, we only change the commutator's oprcom field if
+		 * it has the exact value we expected: InvalidOid when creating an
+		 * operator, or baseId when dropping one.
+		 */
+		if (isDelete && t->oprcom == baseId)
 		{
-			Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);
+			t->oprcom = InvalidOid;
-
+			update_commutator = true;
-			if (!OidIsValid(t->oprcom) || !OidIsValid(t->oprnegate))
+		}
-			{
+		else if (!isDelete && !OidIsValid(t->oprcom))
-				if (!OidIsValid(t->oprnegate))
+		{
-				{
+			t->oprcom = baseId;
-					values[Anum_pg_operator_oprnegate - 1] = ObjectIdGetDatum(baseId);
+			update_commutator = true;
-					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);
-			}
 		}
 
-		heap_close(pg_operator_desc, RowExclusiveLock);
+		/* If any columns were found to need modification, update tuple. */
+		if (update_commutator)
+		{
+			simple_heap_update(pg_operator_desc, &tup->t_self, tup);
+			CatalogUpdateIndexes(pg_operator_desc, tup);
 
-		return;
+			/*
+			 * Do CCI to make the updated tuple visible.  We must do this in
+			 * case the commutator is also the negator.  (Which would be a
+			 * logic error on the operator definer's part, but that's not a
+			 * good reason to fail here.)  We would need a CCI anyway in the
+			 * deletion case for a self-commutator with no negator.
+			 */
+			CommandCounterIncrement();
+		}
 	}
 
-	/* if commutator and negator are different, do two updates */
+	/*
+	 * Similarly find and update the negator, if any.
+	 */
+	if (OidIsValid(negId))
+		tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(negId));
+	else
+		tup = NULL;
 
-	if (HeapTupleIsValid(tup) &&
+	if (HeapTupleIsValid(tup))
-		!(OidIsValid(((Form_pg_operator) GETSTRUCT(tup))->oprcom)))
 	{
-		values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(baseId);
+		Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);
-		replaces[Anum_pg_operator_oprcom - 1] = true;
+		bool		update_negator = false;
 
-		tup = heap_modify_tuple(tup,
+		/*
-								RelationGetDescr(pg_operator_desc),
+		 * Out of due caution, we only change the negator's oprnegate field if
-								values,
+		 * it has the exact value we expected: InvalidOid when creating an
-								nulls,
+		 * operator, or baseId when dropping one.
-								replaces);
+		 */
+		if (isDelete && t->oprnegate == baseId)
+		{
+			t->oprnegate = InvalidOid;
+			update_negator = true;
+		}
+		else if (!isDelete && !OidIsValid(t->oprnegate))
+		{
+			t->oprnegate = baseId;
+			update_negator = true;
+		}
 
-		simple_heap_update(pg_operator_desc, &tup->t_self, tup);
+		/* If any columns were found to need modification, update tuple. */
+		if (update_negator)
+		{
+			simple_heap_update(pg_operator_desc, &tup->t_self, tup);
+			CatalogUpdateIndexes(pg_operator_desc, tup);
 
-		CatalogUpdateIndexes(pg_operator_desc, tup);
+			/*
-
+			 * In the deletion case, do CCI to make the updated tuple visible.
-		values[Anum_pg_operator_oprcom - 1] = (Datum) NULL;
+			 * We must do this in case the operator is its own negator. (Which
-		replaces[Anum_pg_operator_oprcom - 1] = false;
+			 * would be a logic error on the operator definer's part, but
-	}
+			 * that's not a good reason to fail here.)
-
+			 */
-	/* check and update the negator, if necessary */
+			if (isDelete)
-
+				CommandCounterIncrement();
-	tup = SearchSysCacheCopy1(OPEROID, ObjectIdGetDatum(negId));
+		}
-
-	if (HeapTupleIsValid(tup) &&
-		!(OidIsValid(((Form_pg_operator) GETSTRUCT(tup))->oprnegate)))
-	{
-		values[Anum_pg_operator_oprnegate - 1] = ObjectIdGetDatum(baseId);
-		replaces[Anum_pg_operator_oprnegate - 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);
 	}
 
+	/* Close relation and release catalog lock. */
 	heap_close(pg_operator_desc, RowExclusiveLock);
 }
 

src/backend/commands/operatorcmds.c

@@ -341,12 +341,32 @@ RemoveOperatorById(Oid operOid)
 {
 	Relation	relation;
 	HeapTuple	tup;
+	Form_pg_operator op;
 
 	relation = heap_open(OperatorRelationId, RowExclusiveLock);
 
 	tup = SearchSysCache1(OPEROID, ObjectIdGetDatum(operOid));
 	if (!HeapTupleIsValid(tup)) /* should not happen */
 		elog(ERROR, "cache lookup failed for operator %u", operOid);
+	op = (Form_pg_operator) GETSTRUCT(tup);
+
+	/*
+	 * Reset links from commutator and negator, if any.  In case of a
+	 * self-commutator or self-negator, this means we have to re-fetch the
+	 * updated tuple.  (We could optimize away updates on the tuple we're
+	 * about to drop, but it doesn't seem worth convoluting the logic for.)
+	 */
+	if (OidIsValid(op->oprcom) || OidIsValid(op->oprnegate))
+	{
+		OperatorUpd(operOid, op->oprcom, op->oprnegate, true);
+		if (operOid == op->oprcom || operOid == op->oprnegate)
+		{
+			ReleaseSysCache(tup);
+			tup = SearchSysCache1(OPEROID, ObjectIdGetDatum(operOid));
+			if (!HeapTupleIsValid(tup)) /* should not happen */
+				elog(ERROR, "cache lookup failed for operator %u", operOid);
+		}
+	}
 
 	simple_heap_delete(relation, &tup->t_self);
 

src/include/catalog/pg_operator_fn.h

@@ -31,4 +31,6 @@ extern ObjectAddress OperatorCreate(const char *operatorName,
 
 extern ObjectAddress makeOperatorDependencies(HeapTuple tuple, bool isUpdate);
 
+extern void OperatorUpd(Oid baseId, Oid commId, Oid negId, bool isDelete);
+
 #endif   /* PG_OPERATOR_FN_H */

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);

src/test/regress/parallel_schedule

@@ -84,7 +84,7 @@ test: select_into select_distinct select_distinct_on select_implicit select_havi
 # ----------
 # Another group of parallel tests
 # ----------
-test: brin gin gist spgist privileges security_label collate matview lock replica_identity rowsecurity object_address tablesample groupingsets
+test: brin gin gist spgist privileges security_label collate matview lock replica_identity rowsecurity object_address tablesample groupingsets drop_operator
 
 # ----------
 # Another group of parallel tests

src/test/regress/serial_schedule

@@ -89,7 +89,6 @@ test: union
 test: case
 test: join
 test: aggregates
-test: groupingsets
 test: transactions
 ignore: random
 test: random
@@ -114,6 +113,8 @@ test: replica_identity
 test: rowsecurity
 test: object_address
 test: tablesample
+test: groupingsets
+test: drop_operator
 test: alter_generic
 test: alter_operator
 test: misc

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);