Repository: pgjdbc/pgjdbc
Pull request #1329 (2018-11-05 20:04:59)

Allow setFlushedLSN(lsn) and setAppliedLSN(lsn) from outside main loop
Hi,

**TL;DR version**

Marking the fields holding the sequence numbers as `volatile` will allow the API user to call `setFlushedLSN(lsn)` and `setAppliedLSN(lsn)` (and also the getters) from a thread different than the thread doing the main loop. This is useful when doing a blocking `read()`, as it allows another thread to actually acknowledge an event as flushed/applied.

Without this change, one cannot combine `read()` and parallel processing+acknowledging of WAL events, as there is no guarantee that the updated values will be seen by the main replication loop.

**long version**

I'm working on a logical replication consumer implementation which forwards replication event to a queue service (RabbitMQ in this instance). The flow is as following:

1. The main loop consuming the replication events uses `read()` to collect replication events. This is a blocking call, it only returns when a new replication event is read.
2. The replication events, together with their corresponding LSN, are pushed to a `BlockingQueue` (https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html)
3. One or more threads consume this queue and send the replication events to RabbitMQ, following certain policies.
4. From the moment RabbitMQ confirms, we can acknowledge to PostgreSQL, if all previous events were also confirmed by RabbitMQ.

In this flow, replication events are independent from each other. Therefore, it is correct to have multiple active threads in (3) above.

The main replication loop is as following:

    final PGReplicationStream stream = ...;
    final BlockingQueue queue = ...;

    while (true) {
        final ByteBuffer bb = stream.read(); // this blocks
        final Item = new Item(bb, stream.getLastReceivedLSN());

        queue.put(item); // this might also block if the queue is full
    }

The main RabbitMQ loops is/are as following:

    final PGReplicationStream stream = ...; // the same stream as above
    final BlockingQueue queue = ...; // the same queue as above
    final RabbitMQClient client = ...;

    while (true) {
        final Item item = queue.take(); // this might block

        // send to RabbitMQ
        final boolean OK = client.send(item.getBuffer());

        // acknowledge to PostgreSQL
        if (OK && all previous events were also OK) { // this typically happens asynchronously
            stream.setFlushedLSN(item.getLSN());
        }
    }

The change is in this pull request is related to `stream.setFlushedLSN(item.getLSN())` in the code snippet above. Without declaring the LSN members as `volatile`, there is no guarantee that the replication main loop will (ever) notice the update, as it could continue to use its cached value.

pbillen: @davecramer @vlsi Do you see a window to include this in the 42.3 release? It would be great to support the flow as explained. Thanks.
davecramer: @pbillen this isn't a semantic change so I don't see a problem.

That said I think I'd rather synchronize the getters and setters as that seems to be the intent?
pbillen: > @pbillen this isn't a semantic change so I don't see a problem.
> 
> That said I think I'd rather synchronize the getters and setters as that seems to be the intent?

Thank you.

I believe adding `volatile` has the least impact, if any. You could add `synchronized` to the getters and setters, but for the main loop to actually pick up the updated values of the members `lastAppliedLSN` and `lastFlushedLSN` after a setter is called, `V3PGReplicationStream.timeUpdateStatus()` must be synchronized as well.
In addition, for `getLastReceiveLSN()` to become 'visible' outside the main loop (which is beyond the scope of my question but could be useful nonetheless - for example if you want to calculate the current replication lag), you have to synchronize `V3PGReplicationStream.processKeepAliveMessage()` and `V3PGReplicationStream.processXLogData()` as well, to create a memory barrier.

Adding `volatile` avoids all this. Since `V3PGReplicationStream` is by nature a single-threaded implementation (it doesn't make sense to have multiple main loops), you don't have the (minimal) overhead.

That said, it works, of course. So feel free to let me know if you want me to alter the current change!

Thanks again.

Patch:
diff --git a/.github/issue_template.md b/.github/issue_template.md
index a892ae3d..52f04c02 100644
--- a/.github/issue_template.md
+++ b/.github/issue_template.md
@@ -8,11 +8,13 @@
 **Describe the issue**
 A clear and concise description of what the issue is.
 
-**Java Version**
+**Driver Version?** 
 
-**OS Version**
+**Java Version?**
 
-**PostgreSQL Version**
+**OS Version?**
+
+**PostgreSQL Version?**
 
 **To Reproduce**
 Steps to reproduce the behaviour:
diff --git a/.travis/travis_create_slaves.sh b/.travis/travis_create_slaves.sh
index 2eba2450..f772a0ad 100755
--- a/.travis/travis_create_slaves.sh
+++ b/.travis/travis_create_slaves.sh
@@ -3,7 +3,7 @@ set -x -e
 
 if [ -z "$PG_VERSION" ]
 then
-    echo "env PG_VERSION not define";
+    echo "env PG_VERSION not defined";
 elif [ "x${PG_VERSION}" = "xHEAD" ]
 then
     PG_CTL="/usr/local/pgsql/bin/pg_ctl"
@@ -19,11 +19,8 @@ sudo mkdir -p ${PG_SLAVE1_DATADIR}
 sudo chmod 700 ${PG_SLAVE1_DATADIR}
 sudo chown -R postgres:postgres ${PG_SLAVE1_DATADIR}
 
-sudo su postgres -c "$PG_BASEBACKUP -Upostgres -D ${PG_SLAVE1_DATADIR} -X stream"
+sudo su postgres -c "$PG_BASEBACKUP -Upostgres -D ${PG_SLAVE1_DATADIR} -X stream -R"
 
-sudo su postgres -c "echo standby_mode = \'on\' >${PG_SLAVE1_DATADIR}/recovery.conf"
-sudo su postgres -c "echo primary_conninfo = \'host=localhost port=5432 user=test password=test\' >>${PG_SLAVE1_DATADIR}/recovery.conf"
-sudo su postgres -c "echo recovery_target_timeline = \'latest\' >>${PG_SLAVE1_DATADIR}/recovery.conf"
 
 if [[ "x${PG_VERSION}" != "xHEAD" ]]
 then
@@ -38,6 +35,7 @@ then
     sudo sed -i -e "/^[ \t]*hba_file.*/d" ${PG_SLAVE1_DATADIR}/postgresql.conf
     sudo sed -i -e "/^[ \t]*ident_file.*/d" ${PG_SLAVE1_DATADIR}/postgresql.conf
     sudo sed -i -e "s/^#\?hot_standby.*/hot_standby = on/g" ${PG_SLAVE1_DATADIR}/postgresql.conf
+
 fi
 
 #Start Slave 1
@@ -50,25 +48,24 @@ sudo mkdir -p ${PG_SLAVE2_DATADIR}
 sudo chmod 700 ${PG_SLAVE2_DATADIR}
 sudo chown -R postgres:postgres ${PG_SLAVE2_DATADIR}
 
-sudo su postgres -c "$PG_BASEBACKUP -Upostgres -D ${PG_SLAVE2_DATADIR} -X stream"
+sudo su postgres -c "$PG_BASEBACKUP -Upostgres -D ${PG_SLAVE2_DATADIR} -X stream -R"
 
-sudo su postgres -c "echo standby_mode = \'on\' >${PG_SLAVE2_DATADIR}/recovery.conf"
-sudo su postgres -c "echo primary_conninfo = \'host=localhost port=5432 user=test password=test\' >>${PG_SLAVE2_DATADIR}/recovery.conf"
-sudo su postgres -c "echo recovery_target_timeline = \'latest\' >>${PG_SLAVE2_DATADIR}/recovery.conf"
 
 if [[ "x${PG_VERSION}" != "xHEAD" ]]
 then
+
     sudo su postgres -c "echo 'local all all trust' > ${PG_SLAVE2_DATADIR}/pg_hba.conf"
     sudo su postgres -c "echo 'host all all 127.0.0.1/32 trust' >> ${PG_SLAVE2_DATADIR}/pg_hba.conf"
     sudo su postgres -c "echo 'host all all ::1/128 trust' >> ${PG_SLAVE2_DATADIR}/pg_hba.conf"
 
     sudo su postgres -c "touch ${PG_SLAVE2_DATADIR}/pg_ident.conf"
-
-    sudo su postgres -c "cp -f ${PG_DATADIR}/postgresql.conf ${PG_SLAVE2_DATADIR}/postgresql.conf"
+        sudo su postgres -c "cp -f ${PG_DATADIR}/postgresql.conf ${PG_SLAVE2_DATADIR}/postgresql.conf"
     sudo sed -i -e "/^[ \t]*data_directory.*/d" ${PG_SLAVE2_DATADIR}/postgresql.conf
     sudo sed -i -e "/^[ \t]*hba_file.*/d" ${PG_SLAVE2_DATADIR}/postgresql.conf
     sudo sed -i -e "/^[ \t]*ident_file.*/d" ${PG_SLAVE2_DATADIR}/postgresql.conf
     sudo sed -i -e "s/^#\?hot_standby.*/hot_standby = on/g" ${PG_SLAVE2_DATADIR}/postgresql.conf
+
+
 fi
 
 #Start Slave 2
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a3f3865a..7d65a822 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -50,7 +50,7 @@ thrown to caller to be dealt with so no need to log at this verbosity by pgjdbc
 
 ### Fixed
 - getString for PGObject-based types returned "null" string instead of null [PR 1154](https://github.com/pgjdbc/pgjdbc/pull/1154)
-- Field metadata cache can be disabled via databaseMetadataCacheFields=0 [PR 1052](https://github.com/pgjdbc/pgjdbc/pull/1152)
+- Field metadata cache can be disabled via databaseMetadataCacheFields=0 [PR 1052](https://github.com/pgjdbc/pgjdbc/pull/1052)
 - Properly encode special symbols in passwords in BaseDataSource [PR 1201](https://github.com/pgjdbc/pgjdbc/pull/1201)
 - Adjust date, hour, minute, second when rounding nanosecond part of a timestamp [PR 1212](https://github.com/pgjdbc/pgjdbc/pull/1212)
 - perf: reduce memory allocations in query cache [PR 1227](https://github.com/pgjdbc/pgjdbc/pull/1227)
diff --git a/docs/_posts/2018-07-12-42.2.3-release.md b/docs/_posts/2018-07-12-42.2.3-release.md
index 91431d1a..3701f67f 100644
--- a/docs/_posts/2018-07-12-42.2.3-release.md
+++ b/docs/_posts/2018-07-12-42.2.3-release.md
@@ -18,7 +18,7 @@ version: 42.2.3
 
 ### Fixed
 - getString for PGObject-based types returned "null" string instead of null [PR 1154](https://github.com/pgjdbc/pgjdbc/pull/1154)
-- Field metadata cache can be disabled via databaseMetadataCacheFields=0 [PR 1052](https://github.com/pgjdbc/pgjdbc/pull/1152)
+- Field metadata cache can be disabled via databaseMetadataCacheFields=0 [PR 1052](https://github.com/pgjdbc/pgjdbc/pull/1052)
 - Properly encode special symbols in passwords in BaseDataSource [PR 1201](https://github.com/pgjdbc/pgjdbc/pull/1201)
 - Adjust date, hour, minute, second when rounding nanosecond part of a timestamp [PR 1212](https://github.com/pgjdbc/pgjdbc/pull/1212)
 - perf: reduce memory allocations in query cache [PR 1227](https://github.com/pgjdbc/pgjdbc/pull/1227)
diff --git a/pgjdbc/src/main/java/org/postgresql/copy/PGCopyInputStream.java b/pgjdbc/src/main/java/org/postgresql/copy/PGCopyInputStream.java
index 4b4060cc..5bfc100b 100644
--- a/pgjdbc/src/main/java/org/postgresql/copy/PGCopyInputStream.java
+++ b/pgjdbc/src/main/java/org/postgresql/copy/PGCopyInputStream.java
@@ -76,7 +76,7 @@ public class PGCopyInputStream extends InputStream implements CopyOut {
 
   public int read() throws IOException {
     checkClosed();
-    return gotBuf() ? buf[at++] : -1;
+    return gotBuf() ? (buf[at++] & 0xFF)  : -1;
   }
 
   public int read(byte[] buf) throws IOException {
diff --git a/pgjdbc/src/main/java/org/postgresql/core/ServerVersion.java b/pgjdbc/src/main/java/org/postgresql/core/ServerVersion.java
index 5eae078d..9348ba13 100644
--- a/pgjdbc/src/main/java/org/postgresql/core/ServerVersion.java
+++ b/pgjdbc/src/main/java/org/postgresql/core/ServerVersion.java
@@ -24,7 +24,9 @@ public enum ServerVersion implements Version {
   v9_4("9.4.0"),
   v9_5("9.5.0"),
   v9_6("9.6.0"),
-  v10("10")
+  v10("10"),
+  v11("11"),
+  v12("12")
   ;
 
   private final int version;
diff --git a/pgjdbc/src/main/java/org/postgresql/core/v3/replication/V3PGReplicationStream.java b/pgjdbc/src/main/java/org/postgresql/core/v3/replication/V3PGReplicationStream.java
index 38d1afd8..11b4defb 100644
--- a/pgjdbc/src/main/java/org/postgresql/core/v3/replication/V3PGReplicationStream.java
+++ b/pgjdbc/src/main/java/org/postgresql/core/v3/replication/V3PGReplicationStream.java
@@ -35,9 +35,9 @@ public class V3PGReplicationStream implements PGReplicationStream {
   /**
    * Last receive LSN + payload size.
    */
-  private LogSequenceNumber lastReceiveLSN = LogSequenceNumber.INVALID_LSN;
-  private LogSequenceNumber lastAppliedLSN = LogSequenceNumber.INVALID_LSN;
-  private LogSequenceNumber lastFlushedLSN = LogSequenceNumber.INVALID_LSN;
+  private volatile LogSequenceNumber lastReceiveLSN = LogSequenceNumber.INVALID_LSN;
+  private volatile LogSequenceNumber lastAppliedLSN = LogSequenceNumber.INVALID_LSN;
+  private volatile LogSequenceNumber lastFlushedLSN = LogSequenceNumber.INVALID_LSN;
 
   /**
    * @param copyDual         bidirectional copy protocol
diff --git a/pgjdbc/src/main/java/org/postgresql/ds/common/BaseDataSource.java b/pgjdbc/src/main/java/org/postgresql/ds/common/BaseDataSource.java
index 80c1349a..7f78f404 100644
--- a/pgjdbc/src/main/java/org/postgresql/ds/common/BaseDataSource.java
+++ b/pgjdbc/src/main/java/org/postgresql/ds/common/BaseDataSource.java
@@ -1109,6 +1109,9 @@ public abstract class BaseDataSource implements CommonDataSource, Referenceable
 
     Properties p = org.postgresql.Driver.parseURL(url, null);
 
+    if (p == null ) {
+      throw new IllegalArgumentException("URL invalid " + url);
+    }
     for (PGProperty property : PGProperty.values()) {
       if (!this.properties.containsKey(property.getName())) {
         setProperty(property, property.get(p));
diff --git a/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java b/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
index f317b234..f8f53d1b 100644
--- a/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
+++ b/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java
@@ -1066,7 +1066,7 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
     f[13] = new Field("COLUMN_DEF", Oid.VARCHAR);
     f[14] = new Field("SQL_DATA_TYPE", Oid.INT4);
     f[15] = new Field("SQL_DATETIME_SUB", Oid.INT4);
-    f[16] = new Field("CHAR_OCTECT_LENGTH", Oid.INT4);
+    f[16] = new Field("CHAR_OCTET_LENGTH", Oid.INT4);
     f[17] = new Field("ORDINAL_POSITION", Oid.INT4);
     f[18] = new Field("IS_NULLABLE", Oid.VARCHAR);
     f[19] = new Field("SPECIFIC_NAME", Oid.VARCHAR);
@@ -1284,7 +1284,9 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
              + " END "
              + " ELSE NULL "
              + " END "
-             + " AS TABLE_TYPE, d.description AS REMARKS "
+             + " AS TABLE_TYPE, d.description AS REMARKS, "
+             + " '' as TYPE_CAT, '' as TYPE_SCHEM, '' as TYPE_NAME, "
+             + "'' AS SELF_REFERENCING_COL_NAME, '' AS REF_GENERATION "
              + " FROM pg_catalog.pg_namespace n, pg_catalog.pg_class c "
              + " LEFT JOIN pg_catalog.pg_description d ON (c.oid = d.objoid AND d.objsubid = 0) "
              + " LEFT JOIN pg_catalog.pg_class dc ON (d.classoid=dc.oid AND dc.relname='pg_class') "
@@ -1451,7 +1453,7 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
   public ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern,
                               String columnNamePattern) throws SQLException {
 
-    int numberOfFields = 23; // JDBC4
+    int numberOfFields = 24; // JDBC4
     List<byte[][]> v = new ArrayList<byte[][]>(); // The new ResultSet tuple stuff
     Field[] f = new Field[numberOfFields]; // The field descriptors for the new ResultSet
 
@@ -1473,11 +1475,12 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
     f[15] = new Field("CHAR_OCTET_LENGTH", Oid.VARCHAR);
     f[16] = new Field("ORDINAL_POSITION", Oid.INT4);
     f[17] = new Field("IS_NULLABLE", Oid.VARCHAR);
-    f[18] = new Field("SCOPE_CATLOG", Oid.VARCHAR);
+    f[18] = new Field("SCOPE_CATALOG", Oid.VARCHAR);
     f[19] = new Field("SCOPE_SCHEMA", Oid.VARCHAR);
     f[20] = new Field("SCOPE_TABLE", Oid.VARCHAR);
     f[21] = new Field("SOURCE_DATA_TYPE", Oid.INT2);
     f[22] = new Field("IS_AUTOINCREMENT", Oid.VARCHAR);
+    f[23] = new Field( "IS_GENERATED", Oid.VARCHAR);
 
     String sql;
     // a.attnum isn't decremented when preceding columns are dropped,
@@ -1619,6 +1622,8 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
         autoinc = "YES";
       }
       tuple[22] = connection.encodeString(autoinc);
+      // there is no way to tell if the value was actually autogenerated.
+      tuple[23] = connection.encodeString("");
 
       v.add(tuple);
     }
@@ -2650,7 +2655,200 @@ public class PgDatabaseMetaData implements DatabaseMetaData {
   public ResultSet getFunctionColumns(String catalog, String schemaPattern,
       String functionNamePattern, String columnNamePattern)
       throws SQLException {
-    return getProcedureColumns(catalog, schemaPattern, functionNamePattern, columnNamePattern);
+    int columns = 17;
+
+    Field[] f = new Field[columns];
+    List<byte[][]> v = new ArrayList<byte[][]>();
+
+    f[0] = new Field("FUNCTION_CAT", Oid.VARCHAR);
+    f[1] = new Field("FUNCTION_SCHEM", Oid.VARCHAR);
+    f[2] = new Field("FUNCTION_NAME", Oid.VARCHAR);
+    f[3] = new Field("COLUMN_NAME", Oid.VARCHAR);
+    f[4] = new Field("COLUMN_TYPE", Oid.INT2);
+    f[5] = new Field("DATA_TYPE", Oid.INT2);
+    f[6] = new Field("TYPE_NAME", Oid.VARCHAR);
+    f[7] = new Field("PRECISION", Oid.INT2);
+    f[8] = new Field("LENGTH", Oid.INT4);
+    f[9] = new Field("SCALE", Oid.INT2);
+    f[10] = new Field("RADIX", Oid.INT2);
+    f[11] = new Field("NULLABLE", Oid.INT2);
+    f[12] = new Field("REMARKS", Oid.VARCHAR);
+    f[13] = new Field("CHAR_OCTET_LENGTH", Oid.INT4);
+    f[14] = new Field("ORDINAL_POSITION", Oid.INT4);
+    f[15] = new Field("IS_NULLABLE", Oid.VARCHAR);
+    f[16] = new Field("SPECIFIC_NAME", Oid.VARCHAR);
+
+    String sql;
+    sql = "SELECT n.nspname,p.proname,p.prorettype,p.proargtypes, t.typtype,t.typrelid, "
+        + " p.proargnames, p.proargmodes, p.proallargtypes, p.oid "
+        + " FROM pg_catalog.pg_proc p, pg_catalog.pg_namespace n, pg_catalog.pg_type t "
+        + " WHERE p.pronamespace=n.oid AND p.prorettype=t.oid ";
+    if (schemaPattern != null && !schemaPattern.isEmpty()) {
+      sql += " AND n.nspname LIKE " + escapeQuotes(schemaPattern);
+    }
+    if (functionNamePattern != null && !functionNamePattern.isEmpty()) {
+      sql += " AND p.proname LIKE " + escapeQuotes(functionNamePattern);
+    }
+    sql += " ORDER BY n.nspname, p.proname, p.oid::text ";
+
+    byte[] isnullableUnknown = new byte[0];
+
+    Statement stmt = connection.createStatement();
+    ResultSet rs = stmt.executeQuery(sql);
+    while (rs.next()) {
+      byte[] schema = rs.getBytes("nspname");
+      byte[] functionName = rs.getBytes("proname");
+      byte[] specificName =
+          connection.encodeString(rs.getString("proname") + "_" + rs.getString("oid"));
+      int returnType = (int) rs.getLong("prorettype");
+      String returnTypeType = rs.getString("typtype");
+      int returnTypeRelid = (int) rs.getLong("typrelid");
+
+      String strArgTypes = rs.getString("proargtypes");
+      StringTokenizer st = new StringTokenizer(strArgTypes);
+      List<Long> argTypes = new ArrayList<Long>();
+      while (st.hasMoreTokens()) {
+        argTypes.add(Long.valueOf(st.nextToken()));
+      }
+
+      String[] argNames = null;
+      Array argNamesArray = rs.getArray("proargnames");
+      if (argNamesArray != null) {
+        argNames = (String[]) argNamesArray.getArray();
+      }
+
+      String[] argModes = null;
+      Array argModesArray = rs.getArray("proargmodes");
+      if (argModesArray != null) {
+        argModes = (String[]) argModesArray.getArray();
+      }
+
+      int numArgs = argTypes.size();
+
+      Long[] allArgTypes = null;
+      Array allArgTypesArray = rs.getArray("proallargtypes");
+      if (allArgTypesArray != null) {
+        allArgTypes = (Long[]) allArgTypesArray.getArray();
+        numArgs = allArgTypes.length;
+      }
+
+      // decide if we are returning a single column result.
+      if (returnTypeType.equals("b") || returnTypeType.equals("d") || returnTypeType.equals("e")
+          || (returnTypeType.equals("p") && argModesArray == null)) {
+        byte[][] tuple = new byte[columns][];
+        tuple[0] = null;
+        tuple[1] = schema;
+        tuple[2] = functionName;
+        tuple[3] = connection.encodeString("returnValue");
+        tuple[4] = connection
+            .encodeString(Integer.toString(java.sql.DatabaseMetaData.functionReturn));
+        tuple[5] = connection
+            .encodeString(Integer.toString(connection.getTypeInfo().getSQLType(returnType)));
+        tuple[6] = connection.encodeString(connection.getTypeInfo().getPGType(returnType));
+        tuple[7] = null;
+        tuple[8] = null;
+        tuple[9] = null;
+        tuple[10] = null;
+        tuple[11] = connection
+            .encodeString(Integer.toString(java.sql.DatabaseMetaData.functionNullableUnknown));
+        tuple[12] = null;
+        tuple[14] = connection.encodeString(Integer.toString(0));
+        tuple[15] = isnullableUnknown;
+        tuple[16] = specificName;
+
+        v.add(tuple);
+      }
+
+      // Add a row for each argument.
+      for (int i = 0; i < numArgs; i++) {
+        byte[][] tuple = new byte[columns][];
+        tuple[0] = null;
+        tuple[1] = schema;
+        tuple[2] = functionName;
+
+        if (argNames != null) {
+          tuple[3] = connection.encodeString(argNames[i]);
+        } else {
+          tuple[3] = connection.encodeString("$" + (i + 1));
+        }
+
+        int columnMode = DatabaseMetaData.functionColumnIn;
+        if (argModes != null && argModes[i] != null) {
+          if (argModes[i].equals("o")) {
+            columnMode = DatabaseMetaData.functionColumnOut;
+          } else if (argModes[i].equals("b")) {
+            columnMode = DatabaseMetaData.functionColumnInOut;
+          } else if (argModes[i].equals("t")) {
+            columnMode = DatabaseMetaData.functionReturn;
+          }
+        }
+
+        tuple[4] = connection.encodeString(Integer.toString(columnMode));
+
+        int argOid;
+        if (allArgTypes != null) {
+          argOid = allArgTypes[i].intValue();
+        } else {
+          argOid = argTypes.get(i).intValue();
+        }
+
+        tuple[5] =
+            connection.encodeString(Integer.toString(connection.getTypeInfo().getSQLType(argOid)));
+        tuple[6] = connection.encodeString(connection.getTypeInfo().getPGType(argOid));
+        tuple[7] = null;
+        tuple[8] = null;
+        tuple[9] = null;
+        tuple[10] = null;
+        tuple[11] =
+            connection.encodeString(Integer.toString(DatabaseMetaData.functionNullableUnknown));
+        tuple[12] = null;
+        tuple[14] = connection.encodeString(Integer.toString(i + 1));
+        tuple[15] = isnullableUnknown;
+        tuple[16] = specificName;
+
+        v.add(tuple);
+      }
+
+      // if we are returning a multi-column result.
+      if (returnTypeType.equals("c") || (returnTypeType.equals("p") && argModesArray != null)) {
+        String columnsql = "SELECT a.attname,a.atttypid FROM pg_catalog.pg_attribute a "
+            + " WHERE a.attrelid = " + returnTypeRelid
+            + " AND NOT a.attisdropped AND a.attnum > 0 ORDER BY a.attnum ";
+        Statement columnstmt = connection.createStatement();
+        ResultSet columnrs = columnstmt.executeQuery(columnsql);
+        while (columnrs.next()) {
+          int columnTypeOid = (int) columnrs.getLong("atttypid");
+          byte[][] tuple = new byte[columns][];
+          tuple[0] = null;
+          tuple[1] = schema;
+          tuple[2] = functionName;
+          tuple[3] = columnrs.getBytes("attname");
+          tuple[4] = connection
+              .encodeString(Integer.toString(java.sql.DatabaseMetaData.functionColumnResult));
+          tuple[5] = connection
+              .encodeString(Integer.toString(connection.getTypeInfo().getSQLType(columnTypeOid)));
+          tuple[6] = connection.encodeString(connection.getTypeInfo().getPGType(columnTypeOid));
+          tuple[7] = null;
+          tuple[8] = null;
+          tuple[9] = null;
+          tuple[10] = null;
+          tuple[11] = connection
+              .encodeString(Integer.toString(java.sql.DatabaseMetaData.functionNullableUnknown));
+          tuple[12] = null;
+          tuple[14] = connection.encodeString(Integer.toString(0));
+          tuple[15] = isnullableUnknown;
+          tuple[16] = specificName;
+
+          v.add(tuple);
+        }
+        columnrs.close();
+        columnstmt.close();
+      }
+    }
+    rs.close();
+    stmt.close();
+
+    return ((BaseStatement) createMetaDataStatement()).createDriverResultSet(f, v);
   }
 
   public ResultSet getPseudoColumns(String catalog, String schemaPattern, String tableNamePattern,
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/PGReplicationStream.java b/pgjdbc/src/main/java/org/postgresql/replication/PGReplicationStream.java
index 8714c1af..cdb186ae 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/PGReplicationStream.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/PGReplicationStream.java
@@ -12,10 +12,11 @@ import java.nio.ByteBuffer;
 import java.sql.SQLException;
 
 /**
- * Not tread safe replication stream. After complete streaming should be close, for free resource on
- * backend. Periodical status update work only when use {@link PGReplicationStream#read()} method.
- * It means that process wal record should be fast as possible, because during process wal record
- * lead to disconnect by timeout from server.
+ * Not tread safe replication stream (though certain methods can be safely called by different
+ * threads). After complete streaming should be close, for free resource on backend. Periodical
+ * status update work only when use {@link PGReplicationStream#read()} method. It means that
+ * process wal record should be fast as possible, because during process wal record lead to
+ * disconnect by timeout from server.
  */
 public interface PGReplicationStream
     //#if mvn.project.property.postgresql.jdbc.spec >= "JDBC4.1"
@@ -57,7 +58,11 @@ public interface PGReplicationStream
   ByteBuffer readPending() throws SQLException;
 
   /**
-   * Parameter updates by execute {@link PGReplicationStream#read()} method.
+   * <p>Parameter updates by execute {@link PGReplicationStream#read()} method.</p>
+   *
+   * <p>It is safe to call this method in a thread different than the main thread. However, usually this
+   * method is called in the main thread after a successful {@link PGReplicationStream#read()} or
+   * {@link PGReplicationStream#readPending()}, to get the LSN corresponding to the received record.</p>
    *
    * @return not null LSN position that was receive last time via {@link PGReplicationStream#read()}
    *     method
@@ -65,24 +70,31 @@ public interface PGReplicationStream
   LogSequenceNumber getLastReceiveLSN();
 
   /**
-   * Last flushed lsn send in update message to backend. Parameter updates only via {@link
-   * PGReplicationStream#setFlushedLSN(LogSequenceNumber)}
+   * <p>Last flushed lsn send in update message to backend. Parameter updates only via {@link
+   * PGReplicationStream#setFlushedLSN(LogSequenceNumber)}</p>
+   *
+   * <p>It is safe to call this method in a thread different than the main thread.</p>
    *
    * @return not null location of the last WAL flushed to disk in the standby.
    */
   LogSequenceNumber getLastFlushedLSN();
 
   /**
-   * Last applied lsn send in update message to backed. Parameter updates only via {@link
-   * PGReplicationStream#setAppliedLSN(LogSequenceNumber)}
+   * <p>Last applied lsn send in update message to backed. Parameter updates only via {@link
+   * PGReplicationStream#setAppliedLSN(LogSequenceNumber)}</p>
+   *
+   * <p>It is safe to call this method in a thread different than the main thread.</p>
    *
    * @return not null location of the last WAL applied in the standby.
    */
   LogSequenceNumber getLastAppliedLSN();
 
   /**
-   * Set flushed LSN. It parameter will be send to backend on next update status iteration. Flushed
-   * LSN position help backend define which wal can be recycle.
+   * <p>Set flushed LSN. It parameter will be send to backend on next update status iteration. Flushed
+   * LSN position help backend define which wal can be recycle.</p>
+   *
+   * <p>It is safe to call this method in a thread different than the main thread. The updated value
+   * will be sent to the backend in the next status update run.</p>
    *
    * @param flushed not null location of the last WAL flushed to disk in the standby.
    * @see PGReplicationStream#forceUpdateStatus()
@@ -90,8 +102,11 @@ public interface PGReplicationStream
   void setFlushedLSN(LogSequenceNumber flushed);
 
   /**
-   * Parameter used only physical replication and define which lsn already was apply on standby.
-   * Feedback will send to backend on next update status iteration.
+   * <p>Parameter used only physical replication and define which lsn already was apply on standby.
+   * Feedback will send to backend on next update status iteration.</p>
+   *
+   * <p>It is safe to call this method in a thread different than the main thread. The updated value
+   * will be sent to the backend in the next status update run.</p>
    *
    * @param applied not null location of the last WAL applied in the standby.
    * @see PGReplicationStream#forceUpdateStatus()
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/ReplicationSlotInfo.java b/pgjdbc/src/main/java/org/postgresql/replication/ReplicationSlotInfo.java
new file mode 100644
index 00000000..5a8bde42
--- /dev/null
+++ b/pgjdbc/src/main/java/org/postgresql/replication/ReplicationSlotInfo.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2018, PostgreSQL Global Development Group
+ * See the LICENSE file in the project root for more information.
+ */
+
+package org.postgresql.replication;
+
+/**
+ * Information returned on replication slot creation.
+ *
+ * <p>Returned keys of CREATE_REPLICATION_SLOT:
+ * <ol>
+ * <li><b>slot_name</b> String {@code =>} the slot name
+ * <li><b>consistent_point</b> String {@code =>} LSN at which we became consistent
+ * <li><b>snapshot_name</b> String {@code =>} exported snapshot's name (may be <code>null</code>)
+ * <li><b>output_plugin</b> String {@code =>} output plugin (may be <code>null</code>)
+ * </ol>
+ */
+public final class ReplicationSlotInfo {
+
+  private final String slotName;
+  private final ReplicationType replicationType;
+  private final LogSequenceNumber consistentPoint;
+  private final String snapshotName;
+  private final String outputPlugin;
+
+  public ReplicationSlotInfo(String slotName, ReplicationType replicationType,
+      LogSequenceNumber consistentPoint, String snapshotName, String outputPlugin) {
+    this.slotName = slotName;
+    this.replicationType = replicationType;
+    this.consistentPoint = consistentPoint;
+    this.snapshotName = snapshotName;
+    this.outputPlugin = outputPlugin;
+  }
+
+  /**
+   * Replication slot name.
+   *
+   * @return the slot name
+   */
+  public String getSlotName() {
+    return slotName;
+  }
+
+  /**
+   * Replication type of the slot created, might be PHYSICAL or LOGICAL.
+   *
+   * @return ReplicationType, PHYSICAL or LOGICAL
+   */
+  public ReplicationType getReplicationType() {
+    return replicationType;
+  }
+
+  /**
+   * LSN at which we became consistent.
+   *
+   * @return LogSequenceNumber with the consistent_point
+   */
+  public LogSequenceNumber getConsistentPoint() {
+    return consistentPoint;
+  }
+
+  /**
+   * Exported snapshot name at the point of replication slot creation.
+   *
+   * <p>As long as the exporting transaction remains open, other transactions can import its snapshot,
+   * and thereby be guaranteed that they see exactly the same view of the database that the first
+   * transaction sees.
+   *
+   * @return exported snapshot_name (may be <code>null</code>)
+   */
+  public String getSnapshotName() {
+    return snapshotName;
+  }
+
+  /**
+   * Output Plugin used on slot creation.
+   *
+   * @return output_plugin (may be <code>null</code>)
+   */
+  public String getOutputPlugin() {
+    return outputPlugin;
+  }
+
+}
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java
index 4076a4b7..4114befa 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java
@@ -5,13 +5,15 @@
 
 package org.postgresql.replication.fluent;
 
+import org.postgresql.replication.ReplicationSlotInfo;
+
 import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
 
 /**
  * Fluent interface for specify common parameters for create Logical and Physical replication slot.
  */
-public interface ChainedCommonCreateSlotBuilder<T extends ChainedCommonCreateSlotBuilder<T>>  {
+public interface ChainedCommonCreateSlotBuilder<T extends ChainedCommonCreateSlotBuilder<T>> {
 
   /**
    * Replication slots provide an automated way to ensure that the master does not remove WAL
@@ -36,7 +38,9 @@ public interface ChainedCommonCreateSlotBuilder<T extends ChainedCommonCreateSlo
 
   /**
    * Create slot with specified parameters in database.
+   *
+   * @return ReplicationSlotInfo with the information of the created slot.
    * @throws SQLException on error
    */
-  void make() throws SQLException;
+  ReplicationSlotInfo make() throws SQLException;
 }
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java
index 72443c18..27004c82 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java
@@ -6,14 +6,19 @@
 package org.postgresql.replication.fluent.logical;
 
 import org.postgresql.core.BaseConnection;
+import org.postgresql.replication.LogSequenceNumber;
+import org.postgresql.replication.ReplicationSlotInfo;
+import org.postgresql.replication.ReplicationType;
 import org.postgresql.replication.fluent.AbstractCreateSlotBuilder;
 
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 
 public class LogicalCreateSlotBuilder
     extends AbstractCreateSlotBuilder<ChainedLogicalCreateSlotBuilder>
     implements ChainedLogicalCreateSlotBuilder {
+
   private String outputPlugin;
 
   public LogicalCreateSlotBuilder(BaseConnection connection) {
@@ -32,7 +37,7 @@ public class LogicalCreateSlotBuilder
   }
 
   @Override
-  public void make() throws SQLException {
+  public ReplicationSlotInfo make() throws SQLException {
     if (outputPlugin == null || outputPlugin.isEmpty()) {
       throw new IllegalArgumentException(
           "OutputPlugin required parameter for logical replication slot");
@@ -43,15 +48,30 @@ public class LogicalCreateSlotBuilder
     }
 
     Statement statement = connection.createStatement();
+    ResultSet result = null;
+    ReplicationSlotInfo slotInfo = null;
     try {
       statement.execute(String.format(
-              "CREATE_REPLICATION_SLOT %s %s LOGICAL %s",
-              slotName,
-              temporaryOption ? "TEMPORARY" : "",
-              outputPlugin
+          "CREATE_REPLICATION_SLOT %s %s LOGICAL %s",
+          slotName,
+          temporaryOption ? "TEMPORARY" : "",
+          outputPlugin
       ));
+      result = statement.getResultSet();
+      if (result != null && result.next()) {
+        slotInfo = new ReplicationSlotInfo(
+            result.getString("slot_name"),
+            ReplicationType.LOGICAL,
+            LogSequenceNumber.valueOf(result.getString("consistent_point")),
+            result.getString("snapshot_name"),
+            result.getString("output_plugin"));
+      }
     } finally {
+      if (result != null) {
+        result.close();
+      }
       statement.close();
     }
+    return slotInfo;
   }
 }
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java
index 1ae0ff0f..45df2a16 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java
@@ -6,8 +6,12 @@
 package org.postgresql.replication.fluent.physical;
 
 import org.postgresql.core.BaseConnection;
+import org.postgresql.replication.LogSequenceNumber;
+import org.postgresql.replication.ReplicationSlotInfo;
+import org.postgresql.replication.ReplicationType;
 import org.postgresql.replication.fluent.AbstractCreateSlotBuilder;
 
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 
@@ -25,20 +29,35 @@ public class PhysicalCreateSlotBuilder
   }
 
   @Override
-  public void make() throws SQLException {
+  public ReplicationSlotInfo make() throws SQLException {
     if (slotName == null || slotName.isEmpty()) {
       throw new IllegalArgumentException("Replication slotName can't be null");
     }
 
     Statement statement = connection.createStatement();
+    ResultSet result = null;
+    ReplicationSlotInfo slotInfo = null;
     try {
       statement.execute(String.format(
-              "CREATE_REPLICATION_SLOT %s %s PHYSICAL",
-              slotName,
-              temporaryOption ? "TEMPORARY" : ""
+          "CREATE_REPLICATION_SLOT %s %s PHYSICAL",
+          slotName,
+          temporaryOption ? "TEMPORARY" : ""
       ));
+      result = statement.getResultSet();
+      if (result != null && result.next()) {
+        slotInfo = new ReplicationSlotInfo(
+            result.getString("slot_name"),
+            ReplicationType.PHYSICAL,
+            LogSequenceNumber.valueOf(result.getString("consistent_point")),
+            result.getString("snapshot_name"),
+            result.getString("output_plugin"));
+      }
     } finally {
+      if (result != null) {
+        result.close();
+      }
       statement.close();
     }
+    return slotInfo;
   }
 }
diff --git a/pgjdbc/src/main/java/org/postgresql/util/ReaderInputStream.java b/pgjdbc/src/main/java/org/postgresql/util/ReaderInputStream.java
index a313c300..1d6372b2 100644
--- a/pgjdbc/src/main/java/org/postgresql/util/ReaderInputStream.java
+++ b/pgjdbc/src/main/java/org/postgresql/util/ReaderInputStream.java
@@ -113,7 +113,7 @@ public class ReaderInputStream extends InputStream {
     while (res != -1) {
       res = read(oneByte);
       if (res > 0) {
-        return oneByte[0];
+        return (oneByte[0] & 0xFF);
       }
     }
     return -1;


Test patch:
diff --git a/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java b/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java
index 961075d2..5a5aab24 100644
--- a/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java
@@ -6,6 +6,9 @@
 package org.postgresql.replication;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeFalse;
 import static org.junit.Assume.assumeTrue;
@@ -223,6 +226,47 @@ public class ReplicationSlotTest {
     assertThat("Slot should not be temporary by default", result, CoreMatchers.equalTo(false));
   }
 
+  @Test
+  public void testCreateLogicalSlotReturnedInfo() throws Exception {
+    PGConnection pgConnection = (PGConnection) replConnection;
+
+    slotName = "pgjdbc_test_create_logical_replication_slot_info";
+
+    ReplicationSlotInfo info = pgConnection
+        .getReplicationAPI()
+        .createReplicationSlot()
+        .logical()
+        .withSlotName(slotName)
+        .withOutputPlugin("test_decoding")
+        .make();
+
+    assertEquals(slotName, info.getSlotName());
+    assertEquals(ReplicationType.LOGICAL, info.getReplicationType());
+    assertNotNull(info.getConsistentPoint());
+    assertNotNull(info.getSnapshotName());
+    assertEquals("test_decoding", info.getOutputPlugin());
+  }
+
+  @Test
+  public void testCreatePhysicalSlotReturnedInfo() throws Exception {
+    PGConnection pgConnection = (PGConnection) replConnection;
+
+    slotName = "pgjdbc_test_create_physical_replication_slot_info";
+
+    ReplicationSlotInfo info = pgConnection
+        .getReplicationAPI()
+        .createReplicationSlot()
+        .physical()
+        .withSlotName(slotName)
+        .make();
+
+    assertEquals(slotName, info.getSlotName());
+    assertEquals(ReplicationType.PHYSICAL, info.getReplicationType());
+    assertNotNull(info.getConsistentPoint());
+    assertNull(info.getSnapshotName());
+    assertNull(info.getOutputPlugin());
+  }
+
   @Test
   public void testCreateTemporaryLogicalSlotPg10AndHigher()
       throws SQLException {
diff --git a/pgjdbc/src/test/java/org/postgresql/test/TestUtil.java b/pgjdbc/src/test/java/org/postgresql/test/TestUtil.java
index 789736a6..b300749c 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/TestUtil.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/TestUtil.java
@@ -389,19 +389,11 @@ public class TestUtil {
     }
   }
 
-  /*
-   * Helper - creates a test table for use by a test
-   */
-  public static void createTable(Connection con, String table, String columns) throws SQLException {
-    // by default we don't request oids.
-    createTable(con, table, columns, false);
-  }
 
   /*
    * Helper - creates a test table for use by a test
    */
-  public static void createTable(Connection con, String table, String columns, boolean withOids)
-      throws SQLException {
+  public static void createTable(Connection con, String table, String columns) throws SQLException {
     Statement st = con.createStatement();
     try {
       // Drop the table
@@ -410,10 +402,6 @@ public class TestUtil {
       // Now create the table
       String sql = "CREATE TABLE " + table + " (" + columns + ")";
 
-      if (withOids) {
-        sql += " WITH OIDS";
-      }
-
       st.executeUpdate(sql);
     } finally {
       closeQuietly(st);
diff --git a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/DatabaseMetaDataTest.java b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/DatabaseMetaDataTest.java
index 77d8d72a..2538aabb 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/DatabaseMetaDataTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/DatabaseMetaDataTest.java
@@ -23,6 +23,7 @@ import java.sql.Connection;
 import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.sql.Types;
@@ -177,8 +178,16 @@ public class DatabaseMetaDataTest {
     assertEquals("metadatatest", tableName);
     String tableType = rs.getString("TABLE_TYPE");
     assertEquals("TABLE", tableType);
+    assertEquals(rs.findColumn("REMARKS"), 5);
+    assertEquals(rs.findColumn("TYPE_CAT"), 6);
+    assertEquals(rs.findColumn("TYPE_SCHEM"),7);
+    assertEquals(rs.findColumn("TYPE_NAME"), 8);
+    assertEquals(rs.findColumn("SELF_REFERENCING_COL_NAME"), 9);
+    assertEquals(rs.findColumn("REF_GENERATION"), 10);
+
     // There should only be one row returned
     assertTrue("getTables() returned too many rows", rs.next() == false);
+
     rs.close();
 
     rs = dbmd.getColumns("", "", "meta%", "%");
@@ -455,9 +464,21 @@ public class DatabaseMetaDataTest {
   @Test
   public void testColumns() throws SQLException {
     // At the moment just test that no exceptions are thrown KJ
+    String [] metadataColumns = {"TABLE_CAT", "TABLE_SCHEM", "TABLE_NAME", "COLUMN_NAME",
+                                 "DATA_TYPE", "TYPE_NAME", "COLUMN_SIZE", "BUFFER_LENGTH",
+                                 "DECIMAL_DIGITS", "NUM_PREC_RADIX", "NULLABLE", "REMARKS",
+                                 "COLUMN_DEF","SQL_DATA_TYPE","SQL_DATETIME_SUB","CHAR_OCTET_LENGTH",
+                                 "ORDINAL_POSITION", "IS_NULLABLE", "SCOPE_CATALOG", "SCOPE_SCHEMA",
+                                 "SCOPE_TABLE", "SOURCE_DATA_TYPE", "IS_AUTOINCREMENT", "IS_GENERATED"};
+
     DatabaseMetaData dbmd = con.getMetaData();
     assertNotNull(dbmd);
     ResultSet rs = dbmd.getColumns(null, null, "pg_class", null);
+    if ( rs.next() ) {
+      for (int i = 0; i < metadataColumns.length; i++) {
+        assertEquals(i + 1,  rs.findColumn(metadataColumns[i]));
+      }
+    }
     rs.close();
   }
 
@@ -1317,4 +1338,68 @@ public class DatabaseMetaDataTest {
     }
   }
 
+  @Test
+  public void testFunctionColumns() throws SQLException {
+    if (!TestUtil.haveMinimumServerVersion(con, ServerVersion.v8_4)) {
+      return;
+    }
+
+    DatabaseMetaData dbmd = con.getMetaData();
+    ResultSet rs = dbmd.getFunctionColumns(null, null, "f1", null);
+
+    ResultSetMetaData rsmd = rs.getMetaData();
+    assertEquals(17, rsmd.getColumnCount());
+    assertEquals("FUNCTION_CAT", rsmd.getColumnName(1));
+    assertEquals("FUNCTION_SCHEM", rsmd.getColumnName(2));
+    assertEquals("FUNCTION_NAME", rsmd.getColumnName(3));
+    assertEquals("COLUMN_NAME", rsmd.getColumnName(4));
+    assertEquals("COLUMN_TYPE", rsmd.getColumnName(5));
+    assertEquals("DATA_TYPE", rsmd.getColumnName(6));
+    assertEquals("TYPE_NAME", rsmd.getColumnName(7));
+    assertEquals("PRECISION", rsmd.getColumnName(8));
+    assertEquals("LENGTH", rsmd.getColumnName(9));
+    assertEquals("SCALE", rsmd.getColumnName(10));
+    assertEquals("RADIX", rsmd.getColumnName(11));
+    assertEquals("NULLABLE", rsmd.getColumnName(12));
+    assertEquals("REMARKS", rsmd.getColumnName(13));
+    assertEquals("CHAR_OCTET_LENGTH", rsmd.getColumnName(14));
+    assertEquals("ORDINAL_POSITION", rsmd.getColumnName(15));
+    assertEquals("IS_NULLABLE", rsmd.getColumnName(16));
+    assertEquals("SPECIFIC_NAME", rsmd.getColumnName(17));
+
+    assertTrue(rs.next());
+    assertEquals(null, rs.getString(1));
+    assertEquals("public", rs.getString(2));
+    assertEquals("f1", rs.getString(3));
+    assertEquals("returnValue", rs.getString(4));
+    assertEquals(DatabaseMetaData.functionReturn, rs.getInt(5));
+    assertEquals(Types.INTEGER, rs.getInt(6));
+    assertEquals("int4", rs.getString(7));
+    assertEquals(0, rs.getInt(15));
+
+    assertTrue(rs.next());
+    assertEquals(null, rs.getString(1));
+    assertEquals("public", rs.getString(2));
+    assertEquals("f1", rs.getString(3));
+    assertEquals("$1", rs.getString(4));
+    assertEquals(DatabaseMetaData.functionColumnIn, rs.getInt(5));
+    assertEquals(Types.INTEGER, rs.getInt(6));
+    assertEquals("int4", rs.getString(7));
+    assertEquals(1, rs.getInt(15));
+
+    assertTrue(rs.next());
+    assertEquals(null, rs.getString(1));
+    assertEquals("public", rs.getString(2));
+    assertEquals("f1", rs.getString(3));
+    assertEquals("$2", rs.getString(4));
+    assertEquals(DatabaseMetaData.functionColumnIn, rs.getInt(5));
+    assertEquals(Types.VARCHAR, rs.getInt(6));
+    assertEquals("varchar", rs.getString(7));
+    assertEquals(2, rs.getInt(15));
+
+    assertTrue(!rs.next());
+
+    rs.close();
+  }
+
 }
diff --git a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/ResultSetMetaDataTest.java b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/ResultSetMetaDataTest.java
index 85a0b04d..dd417c66 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/ResultSetMetaDataTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/ResultSetMetaDataTest.java
@@ -70,7 +70,7 @@ public class ResultSetMetaDataTest extends BaseTest4 {
   public void setUp() throws Exception {
     super.setUp();
     conn = con;
-    TestUtil.createTable(conn, "rsmd1", "a int primary key, b text, c decimal(10,2)", true);
+    TestUtil.createTable(conn, "rsmd1", "a int primary key, b text, c decimal(10,2)");
     TestUtil.createTable(conn, "rsmd_cache", "a int primary key");
     TestUtil.createTable(conn, "timetest",
         "tm time(3), tmtz timetz, ts timestamp without time zone, tstz timestamp(6) with time zone");
@@ -110,7 +110,7 @@ public class ResultSetMetaDataTest extends BaseTest4 {
   @Test
   public void testStandardResultSet() throws SQLException {
     Statement stmt = conn.createStatement();
-    ResultSet rs = stmt.executeQuery("SELECT a,b,c,a+c as total,oid,b as d FROM rsmd1");
+    ResultSet rs = stmt.executeQuery("SELECT a,b,c,a+c as total, b as d FROM rsmd1");
     runStandardTests(rs.getMetaData());
     rs.close();
     stmt.close();
@@ -121,7 +121,7 @@ public class ResultSetMetaDataTest extends BaseTest4 {
     assumePreparedStatementMetadataSupported();
 
     PreparedStatement pstmt =
-        conn.prepareStatement("SELECT a,b,c,a+c as total,oid,b as d FROM rsmd1 WHERE b = ?");
+        conn.prepareStatement("SELECT a,b,c,a+c as total, b as d FROM rsmd1 WHERE b = ?");
     runStandardTests(pstmt.getMetaData());
     pstmt.close();
   }
@@ -129,15 +129,14 @@ public class ResultSetMetaDataTest extends BaseTest4 {
   private void runStandardTests(ResultSetMetaData rsmd) throws SQLException {
     PGResultSetMetaData pgrsmd = (PGResultSetMetaData) rsmd;
 
-    assertEquals(6, rsmd.getColumnCount());
+    assertEquals(5, rsmd.getColumnCount());
 
     assertEquals("a", rsmd.getColumnLabel(1));
     assertEquals("total", rsmd.getColumnLabel(4));
 
     assertEquals("a", rsmd.getColumnName(1));
-    assertEquals("oid", rsmd.getColumnName(5));
     assertEquals("", pgrsmd.getBaseColumnName(4));
-    assertEquals("b", pgrsmd.getBaseColumnName(6));
+    assertEquals("b", pgrsmd.getBaseColumnName(5));
 
     assertEquals(Types.INTEGER, rsmd.getColumnType(1));
     assertEquals(Types.VARCHAR, rsmd.getColumnType(2));
diff --git a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/UpdateableResultTest.java b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/UpdateableResultTest.java
index 34631605..2a026f7d 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/UpdateableResultTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/UpdateableResultTest.java
@@ -37,8 +37,7 @@ public class UpdateableResultTest extends BaseTest4 {
   public void setUp() throws Exception {
     super.setUp();
     TestUtil.createTable(con, "updateable",
-        "id int primary key, name text, notselected text, ts timestamp with time zone, intarr int[]",
-        true);
+        "id int primary key, name text, notselected text, ts timestamp with time zone, intarr int[]");
     TestUtil.createTable(con, "second", "id1 int primary key, name1 text");
     TestUtil.createTable(con, "stream", "id int primary key, asi text, chr text, bin bytea");
     TestUtil.createTable(con, "multicol", "id1 int not null, id2 int not null, val text");
@@ -330,7 +329,7 @@ public class UpdateableResultTest extends BaseTest4 {
     } catch (SQLException ex) {
     }
 
-    rs = st.executeQuery("select oid,* from updateable");
+    rs = st.executeQuery("select * from updateable");
     assertTrue(rs.first());
     rs.updateInt("id", 3);
     rs.updateString("name", "dave3");
diff --git a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/optional/SimpleDataSourceTest.java b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/optional/SimpleDataSourceTest.java
index 90a7e62f..644d2736 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/optional/SimpleDataSourceTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/optional/SimpleDataSourceTest.java
@@ -5,8 +5,11 @@
 
 package org.postgresql.test.jdbc2.optional;
 
+import org.postgresql.ds.PGSimpleDataSource;
 import org.postgresql.jdbc2.optional.SimpleDataSource;
 
+import org.junit.Test;
+
 /**
  * Performs the basic tests defined in the superclass. Just adds the configuration logic.
  *
@@ -24,4 +27,11 @@ public class SimpleDataSourceTest extends BaseDataSourceTest {
       setupDataSource(bds);
     }
   }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testTypoPostgresUrl() {
+    PGSimpleDataSource ds = new PGSimpleDataSource();
+    // this should fail because the protocol is wrong.
+    ds.setUrl("jdbc:postgres://localhost:5432/test");
+  }
 }
diff --git a/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionParseTest.java b/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionParseTest.java
index 88bbc95c..ffe1a90c 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionParseTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionParseTest.java
@@ -47,6 +47,7 @@ public class ServerVersionParseTest {
         {"9.6", 90600, null},
         {"10", 100000, null},
         {"11", 110000, null},
+        {"12", 120000, null},
         /* Multidigit */
         {"9.4.10", 90410, null},
         {"9.20.10", 92010, null},
diff --git a/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionTest.java b/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionTest.java
index 3e320890..a0a80611 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/util/ServerVersionTest.java
@@ -22,4 +22,14 @@ public class ServerVersionTest {
       prev = serverVersion;
     }
   }
+
+  @Test
+  public void testVersions() {
+    Assert.assertEquals(ServerVersion.v12.getVersionNum(), ServerVersion.from("12.0").getVersionNum());
+    Assert.assertEquals(120004, ServerVersion.from("12.4").getVersionNum());
+    Assert.assertEquals(ServerVersion.v11.getVersionNum(), ServerVersion.from("11.0").getVersionNum());
+    Assert.assertEquals(110006, ServerVersion.from("11.6").getVersionNum());
+    Assert.assertEquals(ServerVersion.v10.getVersionNum(), ServerVersion.from("10.0").getVersionNum());
+    Assert.assertTrue(ServerVersion.v9_6.getVersionNum() < ServerVersion.from("9.6.4").getVersionNum());
+  }
 }