[YSQL] #3979 #4360 Reduce version mismatch errors and improve fault t…

…olerance for DDLs

Summary:
Maintain the ysql catalog version in a (new) system table for new clusters.
The table schema is:
```
            Table "pg_catalog.pg_yb_catalog_version"
        Column         |  Type  | Collation | Nullable | Default
-----------------------+--------+-----------+----------+---------
 db_oid                | oid    |           | not null |
 current_version       | bigint |           | not null |
 last_breaking_version | bigint |           | not null |
Indexes:
    "pg_yb_catalog_version_db_oid_index" PRIMARY KEY,
```

This means we now:
1. Only modify the version number if the overall DDL statement (txn) committed
   successfully (fixing an important correctness bug with version management)
2. Only increment the version number (at most) once per change (minimizing
   version mismatch errors)
- Explicitly distinguish safe vs unsafe changes, so we only need to force a
  refresh during a transaction if necessary (further minimizing version
  mismatch errors).

Additionally, adjust the order in which internal operations are executed
following this design doc:
https://github.com/yugabyte/yugabyte-db/blob/master/architecture/design/online-schema-migrations.md
The goal being to guarantee that under any failure scenario that cluster is
left in a consistent state.

Also added a new (master) test gflag `TEST_ysql_catalog_write_rejection_percentage`
that rejects writes to YSQL system catalog tables with the set percentage (default `0`).

Future work:
1. Implement the upgrade path for old clusters to safely migrate to this new
   versioning method.
2. Maintain the version per-database instead of one per cluster (schema is
   already set up for that).
3. Once more schema changes are handled by the online schema migrations plan,
   adjust their version increment (to become non-breaking) to fully avoid the
   DDL mismatch errors.
4. While all failures should leave the cluster in a consistent state, some may
   leave behind orphan DocDB relations. These are harmless from correctness
   perspective to OID/UUID uniqueness but we should have a simple (perhaps
   automated way to clean them up).
5. Consider how to handle truncate (in particular when used as part of drop colocated table).

Test Plan:
1. Existing jenkins tests
2. Added TestPgDdlFaultTolerance.java
3. Run sysbench with more than one threads.

Reviewers: neha, neil, alex, jason

Reviewed By: alex, jason

Subscribers: alex, vgopalan, mikhail, nicolas, yql, bogdan

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

build-support/lsan-suppressions.txt

@@ -33,5 +33,10 @@ leak:escape_single_quotes_ascii
 leak:gaih_inet
 leak:pg_strdup
 
+# When we push down PG/YSQL expression execution we prepare long-lived, thread-local memory
+# contexts in DocDB to use for evaluating the expressions. These contexts are never deleted, only
+# reset, so LSAN/ASAN reports it as a leak for relevant PG/YSQL tests.
+leak:YbgPrepareMemoryContext
+
 # https://gist.githubusercontent.com/mbautin/015cb594c8281e1afc7ee7b3b5230fce/raw
 leak:__strdup

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

@@ -17,6 +17,7 @@
 import static org.yb.util.SanitizerUtil.isASAN;
 import static org.yb.util.SanitizerUtil.isTSAN;
 
+import com.google.common.net.HostAndPort;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
@@ -33,8 +34,10 @@
 import org.postgresql.util.PSQLException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.yb.client.AsyncYBClient;
 import org.yb.client.IsInitDbDoneResponse;
 import org.yb.client.TestUtils;
+import org.yb.client.YBClient;
 import org.yb.minicluster.*;
 import org.yb.minicluster.Metrics.YSQLStat;
 import org.yb.pgsql.cleaners.ClusterCleaner;
@@ -1024,7 +1027,6 @@ protected void createSimpleTable(Statement statement, String tableName) throws S
   }
 
   /**
-   *
    * @param statement The statement used to execute the query.
    * @param query The query string.
    * @param errorSubstring A (case-insensitive) substring of the expected error message.
@@ -1043,6 +1045,24 @@ protected void runInvalidQuery(Statement statement, String query, String errorSu
     }
   }
 
+  /**
+   * Verify that a (write) query succeeds with a warning matching the given substring.
+   * @param statement The statement used to execute the query.
+   * @param query The query string.
+   * @param warningSubstring A (case-insensitive) substring of the expected warning message.
+   */
+  protected void verifyStatementWarning(Statement statement,
+                                        String query,
+                                        String warningSubstring) throws SQLException {
+    statement.execute(query);
+    SQLWarning warning = statement.getWarnings();
+    assertNotEquals("Expected (at least) one warning", null, warning);
+    assertTrue(String.format("Unexpected Warning Message. Got: '%s', expected to contain : '%s",
+                             warning.getMessage(), warningSubstring),
+               StringUtils.containsIgnoreCase(warning.getMessage(), warningSubstring));
+    assertEquals("Expected (at most) one warning", null, warning.getNextWarning());
+  }
+
   protected String getSimpleTableCreationStatement(
       String tableName,
       String valueColumnName,
@@ -1321,6 +1341,24 @@ protected String getExplainAnalyzeOutput(Statement stmt, String query) throws Ex
   }
 
   /** Immutable connection builder */
+  private void runProcess(String... args) throws Exception {
+    assertEquals(0, new ProcessBuilder(args).start().waitFor());
+  }
+
+  protected HostAndPort getMasterLeaderAddress() {
+    return miniCluster.getClient().getLeaderMasterHostAndPort();
+  }
+
+  protected void setServerFlag(HostAndPort server, String flag, String value) throws Exception {
+    runProcess(TestUtils.findBinary("yb-ts-cli"),
+               "--server_address",
+               server.toString(),
+               "set_flag",
+               "-force",
+               flag,
+               value);
+  }
+
   public static class ConnectionBuilder implements Cloneable {
     private static final int MAX_CONNECTION_ATTEMPTS = 15;
     private static final int INITIAL_CONNECTION_DELAY_MS = 500;

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

@@ -620,6 +620,7 @@ public void testAttributes() throws Exception {
 
       // Cannot create users with unencrypted passwords.
       statement.execute("DROP ROLE IF EXISTS pass_role");
+      waitForTServerHeartbeat();
       runInvalidQuery(
           statement,
           "CREATE ROLE pass_role LOGIN UNENCRYPTED PASSWORD 'pass'",
@@ -633,11 +634,12 @@ public void testAttributes() throws Exception {
       Calendar cal = Calendar.getInstance();
       cal.setTime(new Date());
       cal.add(Calendar.SECOND, 10);
-      Timestamp expiriationTime = new Timestamp(cal.getTime().getTime());
+      Timestamp expirationTime = new Timestamp(cal.getTime().getTime());
 
       statement.execute("DROP ROLE IF EXISTS pass_role");
+      waitForTServerHeartbeat();
       statement.execute("CREATE ROLE pass_role LOGIN PASSWORD 'password' " +
-          "VALID UNTIL '" + expiriationTime.toString() + "'");
+          "VALID UNTIL '" + expirationTime.toString() + "'");
 
       // Can connect now.
       try (Connection ignored = passRoleUserConnBldr.withPassword("password").connect()) {

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

@@ -200,7 +200,7 @@ public void testNoDDLRetry() throws Exception {
          Statement statement2 = connection2.createStatement()) {
       // Create a table with connection 1.
       statement1.execute("CREATE TABLE a(id int primary key)");
-
+      statement1.execute("ALTER TABLE a ADD b int");
       // Create a table with connection 2 (should fail)
       runInvalidQuery(statement2, "CREATE TABLE b(id int primary key)",
           "Catalog Version Mismatch");
@@ -353,16 +353,20 @@ public void testConsistentNonRetryableTransactions() throws Exception {
       // Perform a DDL operation, which cannot (as of 07/01/2019) be rolled back.
       statement2.execute("CREATE TABLE other_table(id int)");
 
+      statement2.execute("SELECT * FROM test_table");
+
       // Modify table from connection 2.
       statement1.execute("ALTER TABLE test_table ADD COLUMN c int");
 
       waitForTServerHeartbeat();
 
-      // Select does not fail, so rollback is not needed.
-      statement2.execute("SELECT * FROM test_table");
+      // Select should fail because the alter modified the table (catalog version mismatch).
+      runInvalidQuery(statement2,"SELECT * FROM test_table", "Catalog Version Mismatch");
+
+      // COMMIT will succeed as a command but will rollback the transaction due to the error above.
       statement2.execute("COMMIT");
 
-      // Check that the table was created.
+      // Check that the other table was created.
       statement2.execute("SELECT * FROM other_table");
     }
   }