Allow StringCodec usage for CITEXT.

We now allow registration of the StringCodec for CITEXT usage through the UNSPECIFIED OID.

[resolves #551]

Signed-off-by: Mark Paluch <mpaluch@vmware.com>

README.md

@@ -277,6 +277,38 @@ The following types are supported for JSON exchange:
 * `String`
 * `InputStream` (must be closed after usage to avoid memory leaks)
 
+## CITEXT support
+
+[CITEXT](https://www.postgresql.org/docs/current/citext.html) is a built-in extension to support case-insensitive `text` columns. By default, the driver sends all string values as `VARCHAR` that cannot be used directly with `CITEXT` (without casting or converting values in your SQL).
+
+If you cast input, then you can send parameters to the server without further customization of the driver:
+
+```sql
+CREATE TABLE test (ci CITEXT);
+SELECT ci FROM test WHERE ci = $1::citext;
+```
+
+If you want to send individual `String`-values in a CITEXT-compatible way, then use `Parameters.in(…)`:
+
+```java
+connection.createStatement("SELECT ci FROM test WHERE ci = $1")
+            .bind("$1", Parameters.in(PostgresqlObjectId.UNSPECIFIED, "Hello"))
+            .execute();
+```
+
+If you do not have control over the created SQL or you want to send all `String` values in a CITEXT-compatible way, then you can customize the driver configuration by registering a `StringCodec` to send `String` values with the `UNSPECIFIED` OID to let Postgres infer the value type from the provided values:
+
+```java
+Builder builder = PostgresqlConnectionConfiguration.builder();
+
+builder.codecRegistrar((connection, allocator, registry) -> {
+    registry.addFirst(new StringCodec(allocator, PostgresqlObjectId.UNSPECIFIED, PostgresqlObjectId.VARCHAR_ARRAY));
+    return Mono.empty();
+});
+```
+
+You can register also the `CodecRegistrar` as [`Extension`](#extension-mechanism) so that it gets auto-detected during `ConnectionFactory` creation.
+
 ## Cursors
 
 The driver can consume cursors that were created by PL/pgSQL as `refcursor`. 

src/main/java/io/r2dbc/postgresql/codec/StringCodec.java

@@ -39,7 +39,7 @@
 import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
 
-final class StringCodec extends AbstractCodec<String> implements ArrayCodecDelegate<String> {
+public final class StringCodec extends AbstractCodec<String> implements ArrayCodecDelegate<String> {
 
     static final Codec<String> STRING_DECODER = StringDecoder.INSTANCE;
 
@@ -53,11 +53,11 @@ final class StringCodec extends AbstractCodec<String> implements ArrayCodecDeleg
 
     private final PostgresTypeIdentifier arrayType;
 
-    StringCodec(ByteBufAllocator byteBufAllocator) {
+    public StringCodec(ByteBufAllocator byteBufAllocator) {
         this(byteBufAllocator, VARCHAR, VARCHAR_ARRAY);
     }
 
-    StringCodec(ByteBufAllocator byteBufAllocator, PostgresTypeIdentifier defaultType, PostgresTypeIdentifier arrayType) {
+    public StringCodec(ByteBufAllocator byteBufAllocator, PostgresTypeIdentifier defaultType, PostgresTypeIdentifier arrayType) {
         super(String.class);
         this.byteBufAllocator = Assert.requireNonNull(byteBufAllocator, "byteBufAllocator must not be null");
         this.defaultType = Assert.requireNonNull(defaultType, "defaultType must not be null");

src/test/java/io/r2dbc/postgresql/codec/StringCodecIntegrationTests.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.r2dbc.postgresql.codec;
+
+import io.r2dbc.postgresql.AbstractIntegrationTests;
+import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
+import io.r2dbc.postgresql.PostgresqlConnectionFactory;
+import io.r2dbc.postgresql.api.PostgresqlConnection;
+import io.r2dbc.spi.Parameters;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+/**
+ * Integration tests for {@link StringCodec} usage with CITEXT and customization options.
+ */
+class StringCodecIntegrationTests extends AbstractIntegrationTests {
+
+    @Test
+    void stringCodecShouldConsiderCIText() {
+
+        SERVER.getJdbcOperations().execute("CREATE EXTENSION IF NOT EXISTS citext WITH SCHEMA public");
+
+        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+        SERVER.getJdbcOperations().execute("CREATE TABLE test (ci CITEXT, cs VARCHAR)");
+        SERVER.getJdbcOperations().execute("INSERT INTO test VALUES('HeLlO', 'HeLlO')");
+
+        this.connection.createStatement("SELECT ci FROM test WHERE ci = $1::citext")
+            .bind("$1", "Hello")
+            .execute()
+            .flatMap(it -> it.map(r -> r.get("ci")))
+            .as(StepVerifier::create)
+            .expectNext("HeLlO")
+            .verifyComplete();
+
+        this.connection.createStatement("SELECT ci FROM test WHERE ci = $1")
+            .bind("$1", Parameters.in(PostgresqlObjectId.UNSPECIFIED, "Hello"))
+            .execute()
+            .flatMap(it -> it.map(r -> r.get("ci")))
+            .as(StepVerifier::create)
+            .expectNext("HeLlO")
+            .verifyComplete();
+
+        this.connection.createStatement("SELECT cs::citext = $1 FROM test")
+            .bind("$1", Parameters.in(PostgresqlObjectId.UNSPECIFIED, "Hello"))
+            .execute()
+            .flatMap(it -> it.map(r -> r.get(0)))
+            .as(StepVerifier::create)
+            .expectNext(true)
+            .verifyComplete();
+
+        this.connection.createStatement("SELECT cs::citext = $1 FROM test")
+            .bind("$1", "Hello")
+            .execute()
+            .flatMap(it -> it.map(r -> r.get(0)))
+            .as(StepVerifier::create)
+            .expectNext(false)
+            .verifyComplete();
+
+        SERVER.getJdbcOperations().execute("DROP TABLE test");
+    }
+
+    @Test
+    void shouldApplyCustomizedCodec() {
+
+        SERVER.getJdbcOperations().execute("CREATE EXTENSION IF NOT EXISTS citext WITH SCHEMA public");
+
+        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+        SERVER.getJdbcOperations().execute("CREATE TABLE test ( ci CITEXT, cs VARCHAR)");
+        SERVER.getJdbcOperations().execute("INSERT INTO test VALUES('HELLO', 'HELLO')");
+
+        PostgresqlConnectionFactory custom = getConnectionFactory(builder -> builder.codecRegistrar((connection1, allocator, registry) -> {
+            registry.addFirst(new StringCodec(allocator, PostgresqlObjectId.UNSPECIFIED, PostgresqlObjectId.VARCHAR_ARRAY));
+            return Mono.empty();
+        }));
+
+        PostgresqlConnection customizedConnection = custom.create().block();
+
+        customizedConnection.createStatement("SELECT cs::citext = $1 FROM test")
+            .bind("$1", "Hello")
+            .execute()
+            .flatMap(it -> it.map(r -> r.get(0)))
+            .as(StepVerifier::create)
+            .expectNext(true)
+            .verifyComplete();
+
+        customizedConnection.close().block();
+
+        SERVER.getJdbcOperations().execute("DROP TABLE test");
+    }
+
+}