#659: [YSQL] Support updating primary key columns

Summary:
# Main changes:

* We now allow `UPDATE` to change primary key columns.
* For that, when we detect that `UPDATE` affects one or more primary key columns, instead of a regular "update", we perform a "replace" operation - we delete an existing row and insert a new one instead.
  - This is less efficient than a normal `UPDATE` (especially in a single row case), so we should recommend avoiding overusing this for performance reasons.
  - Note that unlike a regular "update", this changes `ybctid` of an updated row.
* "Single-row transaction" optimization is not applicable to such "replacing" `UPDATE`s, so it was disabled for them at the query planner level.
* Fixed various broken `pg_regress` tests, but most importantly `yb_pg_foreign_key` - I used original `foreign_key` test as a reference and verified changes against vanilla PostgreSQL v11.

# Other changes:

* `GetYBTablePrimaryKey` was a function where we weirdly used original `FirstLowInvalidHeapAttributeNumber` instead of `YBGetFirstLowInvalidAttributeNumber(rel)` as we do in other places. This has been changed for uniformity.
* A bunch of (sometimes local) utility functions were moved to `pg_yb_utils.h` and `ybcplan.h` since they're used across different source files now. Their names were changed accordingly.
* Avoided unnecessary reopen of relation in `yb_single_row_update_or_delete_path`
* `YBCAllPrimaryKeysProvided` changed to rely on `YBGetTablePrimaryKeyBms`

---

Resolves #659

Test Plan:
ybd --java-test org.yb.pgsql.TestPgUpdatePrimaryKey

Also indirectly by other tests:

ybd --java-test org.yb.pgsql.TestPgUpdate
ybd --java-test org.yb.pgsql.TestPgForeignKey
ybd --java-test org.yb.pgsql.TestPgRegressDml
ybd --java-test org.yb.pgsql.TestPgRegressPlpgsql
ybd --java-test org.yb.pgsql.TestPgRegressAuthorization

Reviewers: neil, mikhail, dmitry, mihnea

Reviewed By: mihnea

Subscribers: yql

Differential Revision: https://phabricator.dev.yugabyte.com/D8998

java/yb-pgsql/src/test/java/org/yb/pgsql/BasePgSQLTest.java

@@ -915,14 +915,25 @@ protected Comparable toComparable(Object obj) {
         "Cannot cast to Comparable " + obj.getClass().getSimpleName() + ": " + obj.toString());
   }
 
+  protected Row getCurrentRow(ResultSet rs) throws SQLException {
+    Comparable[] elems = new Comparable[rs.getMetaData().getColumnCount()];
+    for (int i = 0; i < elems.length; i++) {
+      elems[i] = toComparable(rs.getObject(i + 1)); // Column index starts from 1.
+    }
+    return new Row(elems);
+  }
+
+  protected Row getSingleRow(ResultSet rs) throws SQLException {
+    assertTrue("Result set has no rows", rs.next());
+    Row row = getCurrentRow(rs);
+    assertFalse("Result set has more than one row", rs.next());
+    return row;
+  }
+
   protected List<Row> getRowList(ResultSet rs) throws SQLException {
     List<Row> rows = new ArrayList<>();
     while (rs.next()) {
-      Comparable[] elems = new Comparable[rs.getMetaData().getColumnCount()];
-      for (int i = 0; i < elems.length; i++) {
-        elems[i] = toComparable(rs.getObject(i + 1)); // Column index starts from 1.
-      }
-      rows.add(new Row(elems));
+      rows.add(getCurrentRow(rs));
     }
     return rows;
   }
@@ -950,9 +961,13 @@ protected void assertNoRows(Statement stmt, String query) throws SQLException {
 
   protected void assertNextRow(ResultSet rs, Object... values) throws SQLException {
     assertTrue(rs.next());
-    for (int i = 0; i < values.length; i++) {
-      assertEquals(values[i], rs.getObject(i + 1)); // Column index starts from 1.
+    Comparable[] elems = new Comparable[values.length];
+    for (int i = 0; i < elems.length; i++) {
+      elems[i] = toComparable(values[i]);
     }
+    Row expected = new Row(elems);
+    Row actual = getCurrentRow(rs);
+    assertEquals(expected, actual);
   }
 
   protected void assertOneRow(Statement statement,
@@ -984,21 +999,46 @@ protected void assertRowSet(String stmt, Set<Row> expectedRows) throws SQLExcept
     }
   }
 
-  /** Whether or not this select statement uses index. */
-  protected boolean doesUseIndex(String stmt, String index) throws SQLException {
-    try (Statement statement = connection.createStatement();
-        ResultSet rs = statement.executeQuery("EXPLAIN " + stmt)) {
+  protected void assertRowList(Statement statement,
+                               String query,
+                               List<Row> expectedRows) throws SQLException {
+    try (ResultSet rs = statement.executeQuery(query)) {
+      assertEquals(expectedRows, getRowList(rs));
+    }
+  }
+
+  private boolean isQueryPlanContainsSubstring(Statement stmt, String query, String substring)
+      throws SQLException {
+    try (ResultSet rs = stmt.executeQuery("EXPLAIN " + query)) {
       assert (rs.getMetaData().getColumnCount() == 1); // Expecting one string column.
       while (rs.next()) {
-        if (rs.getString(1).contains("Index Scan using " + index)
-            || rs.getString(1).contains("Index Only Scan using " + index)) {
+        if (rs.getString(1).contains(substring)) {
           return true;
         }
       }
       return false;
     }
   }
 
+  /** Whether or not this select statement uses Index Scan with a given index. */
+  protected boolean isIndexScan(Statement stmt, String query, String index)
+      throws SQLException {
+    return isQueryPlanContainsSubstring(stmt, query, "Index Scan using " + index);
+  }
+
+  /** Whether or not this select statement uses Index Only Scan with a given index. */
+  protected boolean isIndexOnlyScan(Statement stmt, String query, String index)
+      throws SQLException {
+    return isQueryPlanContainsSubstring(stmt, query, "Index Only Scan using " + index);
+  }
+
+  /** Whether or not this select statement uses index. */
+  protected boolean doesUseIndex(String query, String index) throws SQLException {
+    try (Statement stmt = connection.createStatement()) {
+      return isIndexScan(stmt, query, index) || isIndexOnlyScan(stmt, query, index);
+    }
+  }
+
   /**
    * Whether or not this select statement requires filtering by Postgres (i.e. not all
    * conditions can be pushed down to YugaByte).

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgForeignKey.java

@@ -13,9 +13,9 @@
 
 package org.yb.pgsql;
 
-import org.junit.After;
-import org.junit.Ignore;
-import org.junit.Test;
+import static org.yb.AssertionWrappers.*;
+
+import org.junit.*;
 import org.junit.runner.RunWith;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -28,10 +28,6 @@
 import java.util.HashSet;
 import java.util.Set;
 
-import static org.yb.AssertionWrappers.assertEquals;
-import static org.yb.AssertionWrappers.assertFalse;
-import static org.yb.AssertionWrappers.assertTrue;
-
 @RunWith(value=YBTestRunnerNonTsanOnly.class)
 public class TestPgForeignKey extends BasePgSQLTest {
   private static final Logger LOG = LoggerFactory.getLogger(TestPgForeignKey.class);
@@ -169,7 +165,7 @@ private void testForeignKeyConflicts(int pgIsolationLevel) throws Exception {
     }
   }
 
-  // Ensure that foreign key caching maintains data correctness and referential integrity.
+  /** Ensure that foreign key caching maintains data correctness and referential integrity. */
   @Test
   public void testForeignKeyCaching() throws Exception {
     Set<Row> expectedPkRows = new HashSet<>();
@@ -217,30 +213,25 @@ public void testForeignKeyCaching() throws Exception {
       checkRows(statement, "fk_unique", expectedFkUniqueRows);
 
       // Check that deleting referenced row fails the transaction.
-      try {
-        statement.execute("BEGIN");
-        statement.execute("INSERT INTO fk_primary(id, b, a) VALUES(20, 20, 2)");
-        statement.execute("DELETE FROM pk WHERE a=2 AND b=20");
-        statement.execute("COMMIT");
-      } catch (PSQLException e) {
-        assertTrue(e.getMessage().contains("Key (a, b)=(2, 20) is still referenced from table"));
-        statement.execute("ROLLBACK");
-      }
+      statement.execute("BEGIN");
+      statement.execute("INSERT INTO fk_primary(id, b, a) VALUES(20, 20, 2)");
+      runInvalidQuery(statement, "DELETE FROM pk WHERE a=2 AND b=20",
+          "Key (a, b)=(2, 20) is still referenced from table");
+      statement.execute("ROLLBACK");
 
-      // Check that updating referenced row fails the transaction.
-      // TODO: Enable this test case after #3583.
-      if (false) {
-        try {
-          statement.execute("BEGIN");
-          statement.execute("INSERT INTO fk_unique(id, y, x) VALUES(2000, 2000, 200)");
-          statement.execute("UPDATE pk SET x=201, y=2001 WHERE x=200 AND y=2000");
-          statement.execute("COMMIT");
-        } catch (PSQLException e) {
-          assertTrue(
-              e.getMessage().contains("Key (x, y)=(200, 2000) is still referenced from table"));
-          statement.execute("ROLLBACK");
-        }
-      }
+      // Check that updating referenced row fails the transaction, take 1.
+      statement.execute("BEGIN");
+      statement.execute("INSERT INTO fk_primary(id, b, a) VALUES(20, 20, 2)");
+      runInvalidQuery(statement, "UPDATE pk SET a=201, b=2001 WHERE a=2 AND b=20",
+          "Key (a, b)=(2, 20) is still referenced from table");
+      statement.execute("ROLLBACK");
+
+      // Check that updating referenced row fails the transaction, take 2.
+      statement.execute("BEGIN");
+      statement.execute("INSERT INTO fk_unique(id, y, x) VALUES(2000, 2000, 200)");
+      runInvalidQuery(statement, "UPDATE pk SET x=201, y=2001 WHERE x=200 AND y=2000",
+          "Key (x, y)=(200, 2000) is still referenced from table");
+      statement.execute("ROLLBACK");
 
       // Check that deleting unrelated rows in pk table remains unaffected by caching.
       statement.execute("insert into pk(b, a, y, x) values (55, 55, 66, 66)");