Support multiple compression algorithms

Author: knizhnik

doc/src/sgml/libpq.sgml

@@ -1139,8 +1139,11 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname
       <term><literal>compression</literal></term>
       <listitem>
       <para>
-        Request compression of libpq traffic. If server is supporting compression, then all libpq messages send both from client to server and
-        visa versa will be compressed. Right now compression algorithm is hardcoded: is it is either zlib (default), either zstd (if Postgres was
+        Request compression of libpq traffic. Client sends to the server list of compression algorithms, supported by client library.
+        If server supports one of this algorithms, then it acknowledges use of this algorithm and then all libpq messages send both from client to server and
+        visa versa will be compressed. If server is not supporting any of the suggested algorithms, then it replies with 'n' (no compression)
+        message and it is up to the client whether to continue work without compression or report error.
+        Supported compression algorithms are chosen at configure time. Right now two libraries are supported: zlib (default) and zstd (if Postgres was
         configured with --with-zstd option). In both cases streaming mode is used.
       </para>
       </listitem>

doc/src/sgml/protocol.sgml

@@ -97,7 +97,7 @@
     Compression is especial useful for importing/exporting data to/from database using COPY command
     and for replication (both physical and logical). Also compression can reduce server response time
     in case of queries returning large amount of data (for example returning JSON, BLOBs, text,...)
-    Right now compression algorithm is hardcoded: is it is either zlib (default), either zstd (if Postgres was
+    Right now two libraries are supported: zlib (default) and zstd (if Postgres was
     configured with --with-zstd option). In both cases streaming mode is used.
   </para>
 
@@ -272,13 +272,15 @@
      </varlistentry>
 
      <varlistentry>
-      <term>CompressionOk</term>
+      <term>CompressionAck</term>
       <listitem>
        <para>
          Server acknowledge using compression for client-server communication protocol.
          Compression can be requested by client by including "compression" option in connection string.
-         Right now compression algorithm is hardcoded, but in future client and server may negotiate to
-         choose proper compression algorithm.
+         Client sends to the server list of compression algorithms, supported by client library.
+         If server supports one of this algorithms, then it acknowledges use of this algorithm and then all libpq messages send both from client to server and
+         visa versa will be compressed. If server is not supporting any of the suggested algorithms, then it replies with 'n' (no compression)
+         message and it is up to the client whether to continue work without compression or report error.
        </para>
       </listitem>
      </varlistentry>
@@ -3421,7 +3423,7 @@ AuthenticationSASLFinal (B)
 
 <varlistentry>
 <term>
-CompressionOk (B)
+CompressionAck (B)
 </term>
 <listitem>
 <para>
@@ -3433,7 +3435,11 @@ CompressionOk (B)
 </term>
 <listitem>
 <para>
-  Acknowledge use of compression for protocol data. After receiving this message bother server and client are switched to compression mode
+  Acknowledge use of compression for protocol data. Client sends to the server list of compression algorithms, supported by client library.
+  If server supports one of this algorithms, then it responds with CompressionAck with identifier (letter) of first such algorithm.
+  If server is not supporting any of the suggested algorithms, then it replies with 'n' (no compression) algorithm.
+  It is up to the client whether to continue work without compression or report error.
+  After receiving this message with algorithm identifer aother than 'n', both server and client are switched to compression mode
   and exchange compressed messages.
 </para>
 </listitem>
@@ -3454,7 +3460,7 @@ CompressionOk (B)
 </term>
 <listitem>
 <para>
-        Used compression algorithm. Right now the following streaming compression algorithms are supported: 'f' - Facebook zstd, 'z' - zlib.
+        Used compression algorithm. Right now the following streaming compression algorithms are supported: 'f' - Facebook zstd, 'z' - zlib, 'n' - no compresion.
 </para>
 </listitem>
 </varlistentry>
@@ -5888,9 +5894,8 @@ StartupMessage (F)
 </term>
 <listitem>
 <para>
-                        Request compression of libpq traffic. Value can be
-                        <literal>0</literal>, <literal>1</literal>, <literal>true</literal>,
-                        <literal>false</literal>,  <literal>on</literal>, <literal>off.</literal>.
+                        Request compression of libpq traffic. Value is list of compression algorithms supported by client:
+                        <literal>'f'</literal> - Facebook zstd, <literal>'z'</literal> - zlib, <literal>'n'</literal> - no compresion.
                         By default compression is disabled.
 </para>
 </listitem>

src/backend/libpq/pqcomm.c

@@ -199,21 +199,38 @@ WaitEventSet *FeBeWaitSet;
 int
 pq_configure(Port* port)
 {
-	if (port->use_compression)
+	char server_compression_algorithms[ZPQ_MAX_ALGORITHMS];
+	char* client_compression_algorithms = port->compression_algorithms;
+	char compression_algorithm = ZPQ_NO_COMPRESSION;
+	char compression[6] = {'z',0,0,0,5,0}; /* message length = 5 */
+	int rc;
+
+	zpq_get_supported_algorithms(server_compression_algorithms);
+
+	if (client_compression_algorithms)
 	{
-		char compression[6] = {'z',0,0,0,5,0}; /* message length = 5 */
-		int rc;
-		compression[5] = zpq_algorithm();
-		/* Switch on compression at client side */
-		socket_set_nonblocking(false);
-		while ((rc = secure_write(MyProcPort, compression, sizeof(compression))) < 0
-			   && errno == EINTR);
-		if ((size_t)rc != sizeof(compression))
-			return -1;
-
-		/* initialize compression */
-		PqStream = zpq_create((zpq_tx_func)secure_write, (zpq_rx_func)secure_read, MyProcPort);
+		while (*client_compression_algorithms != '\0')
+		{
+			if (strchr(server_compression_algorithms, *client_compression_algorithms))
+			{
+				compression_algorithm = *client_compression_algorithms;
+				break;
+			}
+			client_compression_algorithms += 1;
+		}
 	}
+
+	compression[5] = compression_algorithm;
+	/* Switch on compression at client side */
+	socket_set_nonblocking(false);
+	while ((rc = secure_write(MyProcPort, compression, sizeof(compression))) < 0
+		   && errno == EINTR);
+	if ((size_t)rc != sizeof(compression))
+		return -1;
+
+	/* initialize compression */
+	if (zpq_set_algorithm(compression_algorithm))
+		PqStream = zpq_create((zpq_tx_func)secure_write, (zpq_rx_func)secure_read, MyProcPort);
 	return 0;
 }
 
@@ -256,7 +273,6 @@ pq_init(void)
 					  NULL, NULL);
 	AddWaitEventToSet(FeBeWaitSet, WL_LATCH_SET, -1, MyLatch, NULL);
 	AddWaitEventToSet(FeBeWaitSet, WL_POSTMASTER_DEATH, -1, NULL, NULL);
-
 }
 
 /* --------------------------------

src/backend/postmaster/postmaster.c

@@ -2047,15 +2047,7 @@ ProcessStartupPacket(Port *port, bool SSLdone)
 			else if (strcmp(nameptr, "user") == 0)
 				port->user_name = pstrdup(valptr);
 			else if (strcmp(nameptr, "compression") == 0)
-			{
-				if (!parse_bool(valptr, &port->use_compression))
-					ereport(FATAL,
-							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-							 errmsg("invalid boolean value for parameter \"%s\": \"%s\"",
-									"compression",
-									valptr),
-							 errhint("Valid values are: \"false\", \"off\", 0, \"true\", \"on\", 1.")));
-			}
+				port->compression_algorithms = pstrdup(valptr);
 			else if (strcmp(nameptr, "options") == 0)
 				port->cmdline_options = pstrdup(valptr);
 			else if (strcmp(nameptr, "replication") == 0)