Optimize referential integrity checks (todo item)

diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
new file mode 100644
index 03a974a..107408f
*** a/src/backend/utils/adt/ri_triggers.c
--- b/src/backend/utils/adt/ri_triggers.c
*************** static void ri_BuildQueryKeyFull(RI_Quer
*** 205,215 ****
  static void ri_BuildQueryKeyPkCheck(RI_QueryKey *key,
  						const RI_ConstraintInfo *riinfo,
  						int32 constr_queryno);
! static bool ri_KeysEqual(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  			 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
! static bool ri_AllKeysUnequal(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  				  const RI_ConstraintInfo *riinfo, bool rel_is_pk);
! static bool ri_OneKeyEqual(Relation rel, int column,
  			   HeapTuple oldtup, HeapTuple newtup,
  			   const RI_ConstraintInfo *riinfo, bool rel_is_pk);
  static bool ri_AttributesEqual(Oid eq_opr, Oid typeid,
--- 205,215 ----
  static void ri_BuildQueryKeyPkCheck(RI_QueryKey *key,
  						const RI_ConstraintInfo *riinfo,
  						int32 constr_queryno);
! static bool ri_KeysUnchanged(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  			 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
! static bool ri_AllKeysChanged(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  				  const RI_ConstraintInfo *riinfo, bool rel_is_pk);
! static bool ri_OneKeyUnchanged(Relation rel, int column,
  			   HeapTuple oldtup, HeapTuple newtup,
  			   const RI_ConstraintInfo *riinfo, bool rel_is_pk);
  static bool ri_AttributesEqual(Oid eq_opr, Oid typeid,
*************** RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
*** 932,940 ****
  			}
  
  			/*
! 			 * No need to check anything if old and new keys are equal
  			 */
! 			if (ri_KeysEqual(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowShareLock);
  				return PointerGetDatum(NULL);
--- 932,940 ----
  			}
  
  			/*
! 			 * No need to check anything if old and new keys are unchanged
  			 */
! 			if (ri_KeysUnchanged(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowShareLock);
  				return PointerGetDatum(NULL);
*************** RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
*** 1281,1289 ****
  			}
  
  			/*
! 			 * No need to do anything if old and new keys are equal
  			 */
! 			if (ri_KeysEqual(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowExclusiveLock);
  				return PointerGetDatum(NULL);
--- 1281,1289 ----
  			}
  
  			/*
! 			 * No need to do anything if old and new keys are unchanged
  			 */
! 			if (ri_KeysUnchanged(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowExclusiveLock);
  				return PointerGetDatum(NULL);
*************** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
*** 1646,1654 ****
  			}
  
  			/*
! 			 * No need to check anything if old and new keys are equal
  			 */
! 			if (ri_KeysEqual(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowShareLock);
  				return PointerGetDatum(NULL);
--- 1646,1654 ----
  			}
  
  			/*
! 			 * No need to check anything if old and new keys are unchanged
  			 */
! 			if (ri_KeysUnchanged(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowShareLock);
  				return PointerGetDatum(NULL);
*************** RI_FKey_setnull_upd(PG_FUNCTION_ARGS)
*** 1993,2001 ****
  			}
  
  			/*
! 			 * No need to do anything if old and new keys are equal
  			 */
! 			if (ri_KeysEqual(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowExclusiveLock);
  				return PointerGetDatum(NULL);
--- 1993,2001 ----
  			}
  
  			/*
! 			 * No need to do anything if old and new keys are unchanged
  			 */
! 			if (ri_KeysUnchanged(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowExclusiveLock);
  				return PointerGetDatum(NULL);
*************** RI_FKey_setnull_upd(PG_FUNCTION_ARGS)
*** 2012,2024 ****
  			 * our cached plan, unless the update happens to change all
  			 * columns in the key.	Fortunately, for the most common case of a
  			 * single-column foreign key, this will be true.
- 			 *
- 			 * In case you're wondering, the inequality check works because we
- 			 * know that the old key value has no NULLs (see above).
  			 */
  
  			use_cached_query = (riinfo.confmatchtype == FKCONSTR_MATCH_FULL) ||
! 				ri_AllKeysUnequal(pk_rel, old_row, new_row,
  								  &riinfo, true);
  
  			/*
--- 2012,2021 ----
  			 * our cached plan, unless the update happens to change all
  			 * columns in the key.	Fortunately, for the most common case of a
  			 * single-column foreign key, this will be true.
  			 */
  
  			use_cached_query = (riinfo.confmatchtype == FKCONSTR_MATCH_FULL) ||
! 				ri_AllKeysChanged(pk_rel, old_row, new_row,
  								  &riinfo, true);
  
  			/*
*************** RI_FKey_setnull_upd(PG_FUNCTION_ARGS)
*** 2064,2070 ****
  					 * to changed columns in pk_rel's key
  					 */
  					if (riinfo.confmatchtype == FKCONSTR_MATCH_FULL ||
! 						!ri_OneKeyEqual(pk_rel, i, old_row, new_row,
  										&riinfo, true))
  					{
  						appendStringInfo(&querybuf,
--- 2061,2067 ----
  					 * to changed columns in pk_rel's key
  					 */
  					if (riinfo.confmatchtype == FKCONSTR_MATCH_FULL ||
! 						!ri_OneKeyUnchanged(pk_rel, i, old_row, new_row,
  										&riinfo, true))
  					{
  						appendStringInfo(&querybuf,
*************** RI_FKey_setdefault_upd(PG_FUNCTION_ARGS)
*** 2389,2397 ****
  			}
  
  			/*
! 			 * No need to do anything if old and new keys are equal
  			 */
! 			if (ri_KeysEqual(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowExclusiveLock);
  				return PointerGetDatum(NULL);
--- 2386,2394 ----
  			}
  
  			/*
! 			 * No need to do anything if old and new keys are unchanged
  			 */
! 			if (ri_KeysUnchanged(pk_rel, old_row, new_row, &riinfo, true))
  			{
  				heap_close(fk_rel, RowExclusiveLock);
  				return PointerGetDatum(NULL);
*************** RI_FKey_setdefault_upd(PG_FUNCTION_ARGS)
*** 2443,2449 ****
  					 * to changed columns in pk_rel's key
  					 */
  					if (riinfo.confmatchtype == FKCONSTR_MATCH_FULL ||
! 						!ri_OneKeyEqual(pk_rel, i, old_row, new_row,
  										&riinfo, true))
  					{
  						appendStringInfo(&querybuf,
--- 2440,2446 ----
  					 * to changed columns in pk_rel's key
  					 */
  					if (riinfo.confmatchtype == FKCONSTR_MATCH_FULL ||
! 						!ri_OneKeyUnchanged(pk_rel, i, old_row, new_row,
  										&riinfo, true))
  					{
  						appendStringInfo(&querybuf,
*************** RI_FKey_keyequal_upd_pk(Trigger *trigger
*** 2540,2547 ****
  	{
  		case FKCONSTR_MATCH_UNSPECIFIED:
  		case FKCONSTR_MATCH_FULL:
! 			/* Return true if keys are equal */
! 			return ri_KeysEqual(pk_rel, old_row, new_row, &riinfo, true);
  
  			/* Handle MATCH PARTIAL set null delete. */
  		case FKCONSTR_MATCH_PARTIAL:
--- 2537,2544 ----
  	{
  		case FKCONSTR_MATCH_UNSPECIFIED:
  		case FKCONSTR_MATCH_FULL:
! 			/* Return true if keys are unchanged */
! 			return ri_KeysUnchanged(pk_rel, old_row, new_row, &riinfo, true);
  
  			/* Handle MATCH PARTIAL set null delete. */
  		case FKCONSTR_MATCH_PARTIAL:
*************** RI_FKey_keyequal_upd_fk(Trigger *trigger
*** 2585,2592 ****
  	{
  		case FKCONSTR_MATCH_UNSPECIFIED:
  		case FKCONSTR_MATCH_FULL:
! 			/* Return true if keys are equal */
! 			return ri_KeysEqual(fk_rel, old_row, new_row, &riinfo, false);
  
  			/* Handle MATCH PARTIAL set null delete. */
  		case FKCONSTR_MATCH_PARTIAL:
--- 2582,2589 ----
  	{
  		case FKCONSTR_MATCH_UNSPECIFIED:
  		case FKCONSTR_MATCH_FULL:
! 			/* Return true if keys are unchanged */
! 			return ri_KeysUnchanged(fk_rel, old_row, new_row, &riinfo, false);
  
  			/* Handle MATCH PARTIAL set null delete. */
  		case FKCONSTR_MATCH_PARTIAL:
*************** ri_HashPreparedPlan(RI_QueryKey *key, SP
*** 3763,3775 ****
  
  
  /* ----------
!  * ri_KeysEqual -
   *
!  *	Check if all key values in OLD and NEW are equal.
   * ----------
   */
  static bool
! ri_KeysEqual(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  			 const RI_ConstraintInfo *riinfo, bool rel_is_pk)
  {
  	TupleDesc	tupdesc = RelationGetDescr(rel);
--- 3760,3772 ----
  
  
  /* ----------
!  * ri_KeysUnchanged -
   *
!  *	Check if all key values in OLD and NEW are unchanged.
   * ----------
   */
  static bool
! ri_KeysUnchanged(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  			 const RI_ConstraintInfo *riinfo, bool rel_is_pk)
  {
  	TupleDesc	tupdesc = RelationGetDescr(rel);
*************** ri_KeysEqual(Relation rel, HeapTuple old
*** 3792,3812 ****
  	{
  		Datum		oldvalue;
  		Datum		newvalue;
! 		bool		isnull;
! 
! 		/*
! 		 * Get one attribute's oldvalue. If it is NULL - they're not equal.
! 		 */
! 		oldvalue = SPI_getbinval(oldtup, tupdesc, attnums[i], &isnull);
! 		if (isnull)
! 			return false;
  
  		/*
! 		 * Get one attribute's newvalue. If it is NULL - they're not equal.
  		 */
! 		newvalue = SPI_getbinval(newtup, tupdesc, attnums[i], &isnull);
! 		if (isnull)
  			return false;
  
  		/*
  		 * Compare them with the appropriate equality operator.
--- 3789,3809 ----
  	{
  		Datum		oldvalue;
  		Datum		newvalue;
! 		bool		isoldnull;
! 		bool		isnewnull;
  
  		/*
! 		 * Get one attribute's oldvalue and newvalue. If they are both NULL,
! 		 * the attribute is unchanged.  If only one is NULL, it has changed.
  		 */
! 		oldvalue = SPI_getbinval(oldtup, tupdesc, attnums[i], &isoldnull);
! 		newvalue = SPI_getbinval(newtup, tupdesc, attnums[i], &isnewnull);
! 		if (isoldnull || isnewnull)
! 		{
! 			if (isoldnull && isnewnull)
! 				continue;
  			return false;
+ 		}
  
  		/*
  		 * Compare them with the appropriate equality operator.
*************** ri_KeysEqual(Relation rel, HeapTuple old
*** 3821,3833 ****
  
  
  /* ----------
!  * ri_AllKeysUnequal -
   *
!  *	Check if all key values in OLD and NEW are not equal.
   * ----------
   */
  static bool
! ri_AllKeysUnequal(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  				  const RI_ConstraintInfo *riinfo, bool rel_is_pk)
  {
  	TupleDesc	tupdesc = RelationGetDescr(rel);
--- 3818,3830 ----
  
  
  /* ----------
!  * ri_AllKeysChanged -
   *
!  *	Check if all key values in OLD and NEW are changed.
   * ----------
   */
  static bool
! ri_AllKeysChanged(Relation rel, HeapTuple oldtup, HeapTuple newtup,
  				  const RI_ConstraintInfo *riinfo, bool rel_is_pk)
  {
  	TupleDesc	tupdesc = RelationGetDescr(rel);
*************** ri_AllKeysUnequal(Relation rel, HeapTupl
*** 3850,3877 ****
  	{
  		Datum		oldvalue;
  		Datum		newvalue;
! 		bool		isnull;
! 
! 		/*
! 		 * Get one attribute's oldvalue. If it is NULL - they're not equal.
! 		 */
! 		oldvalue = SPI_getbinval(oldtup, tupdesc, attnums[i], &isnull);
! 		if (isnull)
! 			continue;
  
  		/*
! 		 * Get one attribute's newvalue. If it is NULL - they're not equal.
  		 */
! 		newvalue = SPI_getbinval(newtup, tupdesc, attnums[i], &isnull);
! 		if (isnull)
  			continue;
  
  		/*
  		 * Compare them with the appropriate equality operator.
  		 */
  		if (ri_AttributesEqual(eq_oprs[i], RIAttType(rel, attnums[i]),
  							   oldvalue, newvalue))
! 			return false;		/* found two equal items */
  	}
  
  	return true;
--- 3847,3874 ----
  	{
  		Datum		oldvalue;
  		Datum		newvalue;
! 		bool		isoldnull;
! 		bool		isnewnull;
  
  		/*
! 		 * Get one attribute's oldvalue and newvalue. If they are both NULL,
! 		 * the attribute is unchanged.  If only one is NULL, it has changed.
  		 */
! 		oldvalue = SPI_getbinval(oldtup, tupdesc, attnums[i], &isoldnull);
! 		newvalue = SPI_getbinval(newtup, tupdesc, attnums[i], &isnewnull);
! 		if (isoldnull || isnewnull)
! 		{
! 			if (isoldnull && isnewnull)
! 				return false;	/* found two unchanged items */
  			continue;
+ 		}
  
  		/*
  		 * Compare them with the appropriate equality operator.
  		 */
  		if (ri_AttributesEqual(eq_oprs[i], RIAttType(rel, attnums[i]),
  							   oldvalue, newvalue))
! 			return false;		/* found two unchanged items */
  	}
  
  	return true;
*************** ri_AllKeysUnequal(Relation rel, HeapTupl
*** 3879,3895 ****
  
  
  /* ----------
!  * ri_OneKeyEqual -
   *
!  *	Check if one key value in OLD and NEW is equal.  Note column is indexed
   *	from zero.
   *
!  *	ri_KeysEqual could call this but would run a bit slower.  For
   *	now, let's duplicate the code.
   * ----------
   */
  static bool
! ri_OneKeyEqual(Relation rel, int column, HeapTuple oldtup, HeapTuple newtup,
  			   const RI_ConstraintInfo *riinfo, bool rel_is_pk)
  {
  	TupleDesc	tupdesc = RelationGetDescr(rel);
--- 3876,3892 ----
  
  
  /* ----------
!  * ri_OneKeyUnchanged -
   *
!  *	Check if one key value in OLD and NEW is unchanged.  Note column is indexed
   *	from zero.
   *
!  *	ri_KeysUnchanged could call this but would run a bit slower.  For
   *	now, let's duplicate the code.
   * ----------
   */
  static bool
! ri_OneKeyUnchanged(Relation rel, int column, HeapTuple oldtup, HeapTuple newtup,
  			   const RI_ConstraintInfo *riinfo, bool rel_is_pk)
  {
  	TupleDesc	tupdesc = RelationGetDescr(rel);
*************** ri_OneKeyEqual(Relation rel, int column,
*** 3897,3903 ****
  	const Oid  *eq_oprs;
  	Datum		oldvalue;
  	Datum		newvalue;
! 	bool		isnull;
  
  	if (rel_is_pk)
  	{
--- 3894,3901 ----
  	const Oid  *eq_oprs;
  	Datum		oldvalue;
  	Datum		newvalue;
! 	bool		isoldnull;
! 	bool		isnewnull;
  
  	if (rel_is_pk)
  	{
*************** ri_OneKeyEqual(Relation rel, int column,
*** 3911,3928 ****
  	}
  
  	/*
! 	 * Get one attribute's oldvalue. If it is NULL - they're not equal.
! 	 */
! 	oldvalue = SPI_getbinval(oldtup, tupdesc, attnums[column], &isnull);
! 	if (isnull)
! 		return false;
! 
! 	/*
! 	 * Get one attribute's newvalue. If it is NULL - they're not equal.
  	 */
! 	newvalue = SPI_getbinval(newtup, tupdesc, attnums[column], &isnull);
! 	if (isnull)
  		return false;
  
  	/*
  	 * Compare them with the appropriate equality operator.
--- 3909,3925 ----
  	}
  
  	/*
! 	 * Get one attribute's oldvalue and newvalue. If they are both NULL,
! 	 * the attribute is unchanged.  If only one is NULL, it has changed.
  	 */
! 	oldvalue = SPI_getbinval(oldtup, tupdesc, attnums[column], &isoldnull);
! 	newvalue = SPI_getbinval(newtup, tupdesc, attnums[column], &isnewnull);
! 	if (isoldnull || isnewnull)
! 	{
! 		if (isoldnull && isnewnull)
! 			return true;
  		return false;
+ 	}
  
  	/*
  	 * Compare them with the appropriate equality operator.