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]>

src/bin/pgcopydb/ld_transform.c

@@ -2227,12 +2227,15 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 						  update->table.relname);
 		int pos = 0;
 
+		int skippedColumns = 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++)
 			{
@@ -2256,6 +2259,20 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 				 */
 				bool skip = false;
 
+				/*
+				 * 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;
+				 */
 				for (int oc = 0; oc < old->attributes.count; oc++)
 				{
 					LogicalMessageAttribute *oldAttr = &(old->attributes.array[oc]);
@@ -2273,7 +2290,11 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 					}
 				}
 
-				if (!skip)
+				if (skip)
+				{
+					++skippedColumns;
+				}
+				else
 				{
 					if (attr->isgenerated)
 					{
@@ -2360,20 +2381,30 @@ stream_write_update(FILE *out, LogicalMessageUpdate *update)
 			return false;
 		}
 
-		uint32_t hash = hashlittle(buf->data, buf->len, 5381);
+		bool skipUpdate = (skippedColumns == new->attributes.count);
 
-		FFORMAT(out, "PREPARE %x AS %s;\n", hash, buf->data);
+		if (skipUpdate)
+		{
+			log_warn("Skipping UPDATE statement as all columns are "
+					 "the same as the old");
+		}
+		else
+		{
+			uint32_t hash = hashlittle(buf->data, buf->len, 5381);
 
-		destroyPQExpBuffer(buf);
+			FFORMAT(out, "PREPARE %x AS %s;\n", hash, buf->data);
 
-		/*
-		 * Second, the EXECUTE part.
-		 */
-		char *serialized_string = json_serialize_to_string(js);
+			destroyPQExpBuffer(buf);
 
-		FFORMAT(out, "EXECUTE %x%s;\n", hash, serialized_string);
+			/*
+			 * Second, the EXECUTE part.
+			 */
+			char *serialized_string = json_serialize_to_string(js);
 
-		json_free_serialized_string(serialized_string);
+			FFORMAT(out, "EXECUTE %x%s;\n", hash, serialized_string);
+
+			json_free_serialized_string(serialized_string);
+		}
 	}
 
 	return true;