Fix invalid UPDATE stmt when none of the columns change

We need at least two columns to be able to skip a column
if the value is the same in the old and new rows.

Otherwise, we would end up with an invalid UPDATE statement
like below:
```
UPDATE table SET WHERE "id" = 1;
```

Usually, the above could happen when REPLICA IDENTITY is set
to FULL, and the UPDATE statement executed with the same
values as the old ones.
For e.g.
```
UPDATE table SET "id" = 1 WHERE "id" = 1;
```

Solution: Skip the update when all columns in SET clause is equal to the WHERE clause.

Signed-off-by: Arunprasad Rajkumar <[email protected]>

Author: arajkumar

src/bin/pgcopydb/ld_transform.c

@@ -2227,12 +2227,15 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 						  update->table.relname);
 		int pos = 0;
 
+		int skipColumnCount = 0;
+
 		for (int r = 0; r < new->values.count; r++)
 		{
 			LogicalMessageValues *values = &(new->values.array[r]);
 
 			bool first = true;
 
+
 			/* now loop over column values for this VALUES row */
 			for (int v = 0; v < values->cols; v++)
 			{
@@ -2273,7 +2276,11 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 					}
 				}
 
-				if (!skip)
+				if (skip)
+				{
+					++skipColumnCount;
+				}
+				else
 				{
 					if (attr->isgenerated)
 					{
@@ -2360,6 +2367,31 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 			return false;
 		}
 
+		/*
+		 * When all column values in the SET clause are equal to those in the
+		 * WHERE clause, we remove all columns from the SET clause. This results
+		 * in an invalid UPDATE statement like the one shown below:
+		 *
+		 * UPDATE table SET WHERE "id" = 1;
+		 *
+		 * Usually, the above could happen when REPLICA IDENTITY is set to FULL,
+		 * and the UPDATE statement executed with the same values as the old ones.
+		 * For e.g.
+		 * UPDATE table SET "id" = 1 WHERE "id" = 1;
+		 *
+		 * Skip the UPDATE statement in such cases.
+		 */
+		bool skipUpdate = (skipColumnCount == new->attributes.count);
+
+		if (skipUpdate)
+		{
+			log_warn("Skipping UPDATE statement as all columns are "
+					 "the same as the old");
+
+			destroyPQExpBuffer(buf);
+			return true;
+		}
+
 		uint32_t hash = hashlittle(buf->data, buf->len, 5381);
 
 		FFORMAT(out, "PREPARE %x AS %s;\n", hash, buf->data);