Introduce codec mapping caches

pom.xml

@@ -41,8 +41,8 @@
         <junit.version>5.7.0</junit.version>
         <jmh.version>1.32</jmh.version>
         <mbr.version>0.2.0.RELEASE</mbr.version>
-        <logback.version>1.2.3</logback.version>
-        <mockito.version>3.11.2</mockito.version>
+        <logback.version>1.2.5</logback.version>
+        <mockito.version>3.12.4</mockito.version>
         <netty.version>4.1.66.Final</netty.version>
         <postgresql.version>42.2.23</postgresql.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -210,7 +210,7 @@
         </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
-            <artifactId>mockito-core</artifactId>
+            <artifactId>mockito-junit-jupiter</artifactId>
             <version>${mockito.version}</version>
             <scope>test</scope>
         </dependency>

src/jmh/java/io/r2dbc/postgresql/CodecRegistryBenchmarks.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2021 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;
+
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.r2dbc.postgresql.codec.CodecFinderCacheImpl;
+import io.r2dbc.postgresql.codec.CodecFinderDefaultImpl;
+import io.r2dbc.postgresql.codec.Codecs;
+import io.r2dbc.postgresql.codec.DefaultCodecs;
+import io.r2dbc.postgresql.util.ByteBufUtils;
+import org.junit.platform.commons.annotation.Testable;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.time.LocalDateTime;
+import java.util.concurrent.TimeUnit;
+
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.FLOAT4;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.FLOAT8;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.FLOAT8_ARRAY;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.INT2;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.INT2_ARRAY;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.INT4;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.INT4_ARRAY;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.TIMESTAMP;
+import static io.r2dbc.postgresql.codec.PostgresqlObjectId.VARCHAR;
+import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
+import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
+import static io.r2dbc.postgresql.util.TestByteBufAllocator.TEST;
+
+/**
+ * Benchmarks for codec encoding and decoding using cached enabled or disabled registries.
+ */
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Testable
+public class CodecRegistryBenchmarks extends BenchmarkSettings {
+
+    @State(Scope.Benchmark)
+    public static class CodecRegistryHolder {
+
+        @Param({"10", "100", "1000"})
+        public int iterations;
+
+        final ByteBufAllocator byteBufAllocator = new UnpooledByteBufAllocator(false, true);
+
+        DefaultCodecs cacheEnabledRegistry = new DefaultCodecs(byteBufAllocator, false, new CodecFinderCacheImpl());
+
+        DefaultCodecs cacheDisabledRegistry = new DefaultCodecs(byteBufAllocator, false, new CodecFinderDefaultImpl());
+
+    }
+
+    private void decode(Codecs codecs, int iterations, Blackhole voodoo) {
+        for (int i = 0; i < iterations; i++) {
+            voodoo.consume(codecs.decode(
+                TEST.buffer(4).writeInt(200), INT4.getObjectId(), FORMAT_BINARY, Integer.class));
+            voodoo.consume(codecs.decode(
+                ByteBufUtils.encode(TEST, "100"), INT2.getObjectId(), FORMAT_TEXT, Short.class));
+            voodoo.consume(codecs.decode(
+                ByteBufUtils.encode(TEST, "-125.369"), FLOAT8.getObjectId(), FORMAT_TEXT, Double.class));
+            voodoo.consume(codecs.decode(
+                TEST.buffer(4).writeFloat(-65.369f), FLOAT4.getObjectId(), FORMAT_BINARY, Float.class));
+            voodoo.consume(
+                codecs.decode(
+                    ByteBufUtils.encode(TEST, "test"),
+                    VARCHAR.getObjectId(),
+                    FORMAT_TEXT,
+                    String.class));
+            voodoo.consume(
+                codecs.decode(
+                    ByteBufUtils.encode(TEST, "2018-11-04 15:35:00.847108"),
+                    TIMESTAMP.getObjectId(),
+                    FORMAT_TEXT,
+                    LocalDateTime.class));
+            voodoo.consume(codecs.decode(ByteBufUtils.encode(TEST, "{100,200}"), INT2_ARRAY.getObjectId(), FORMAT_TEXT, Object.class));
+            voodoo.consume(codecs.decode(ByteBufUtils.encode(TEST, "{100,200}"), INT4_ARRAY.getObjectId(), FORMAT_TEXT, Object.class));
+            voodoo.consume(codecs.decode(ByteBufUtils.encode(TEST, "{100.5,200.8}"), FLOAT8_ARRAY.getObjectId(), FORMAT_TEXT, Object.class));
+        }
+    }
+
+    @Benchmark
+    public void decodeWithCacheEnabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        decode(holder.cacheEnabledRegistry, holder.iterations, voodoo);
+    }
+
+    @Benchmark
+    public void decodeWithCacheDisabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        decode(holder.cacheDisabledRegistry, holder.iterations, voodoo);
+    }
+
+    private void encode(Codecs codecs, int iterations, Blackhole voodoo) {
+        for (int i = 0; i < iterations; i++) {
+            voodoo.consume(codecs.encode((short) 12));
+            voodoo.consume(codecs.encode(35698));
+            voodoo.consume(codecs.encode(-256.3698));
+            voodoo.consume(codecs.encode(85.7458f));
+            voodoo.consume(codecs.encode("A text value"));
+            voodoo.consume(codecs.encode(LocalDateTime.now()));
+            voodoo.consume(codecs.encode(new Long[]{100L, 200L}));
+            voodoo.consume(codecs.encode(new Double[]{100.5, 200.25}));
+            voodoo.consume(codecs.encodeNull(Integer.class));
+            voodoo.consume(codecs.encodeNull(String.class));
+            voodoo.consume(codecs.encodeNull(Double[].class));
+        }
+    }
+
+    @Benchmark
+    public void encodeWithCacheEnabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        encode(holder.cacheEnabledRegistry, holder.iterations, voodoo);
+    }
+
+    @Benchmark
+    public void encodeWithCacheDisabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        encode(holder.cacheDisabledRegistry, holder.iterations, voodoo);
+    }
+
+}

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

@@ -25,6 +25,7 @@
 import io.r2dbc.postgresql.util.ByteBufUtils;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
 import java.util.Objects;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -74,6 +75,11 @@ public PostgresTypeIdentifier getArrayDataType() {
         return BYTEA_ARRAY;
     }
 
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return Collections.singleton(BYTEA);
+    }
+
     byte[] decode(Format format, ByteBuf byteBuf) {
         if (format == FORMAT_TEXT) {
             return decodeFromHex(byteBuf);

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

@@ -25,9 +25,12 @@
 import reactor.core.publisher.Mono;
 import reactor.util.annotation.Nullable;
 
+import java.util.EnumSet;
 import java.util.function.Supplier;
 
 import static io.r2dbc.postgresql.client.EncodedParameter.NULL_VALUE;
+import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
+import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
 
 /**
  * Abstract codec class that provides a basis for all concrete
@@ -97,8 +100,8 @@ public EncodedParameter encode(Object value, int dataType) {
         return doEncode((T) value, getDataType(dataType));
     }
 
-    PostgresTypeIdentifier getDataType(int dataType) {
-        return PostgresqlObjectId.isValid(dataType) ? PostgresqlObjectId.valueOf(dataType) : new SimplePostgresTypeIdentifier(dataType);
+    public static PostgresTypeIdentifier getDataType(int dataType) {
+        return PostgresqlObjectId.isValid(dataType) ? PostgresqlObjectId.valueOf(dataType) : () -> dataType;
     }
 
     public EncodedParameter encodeNull(int dataType) {
@@ -110,6 +113,12 @@ public Class<?> type() {
         return this.type;
     }
 
+    @Override
+    public Iterable<Format> getFormats() {
+        // Unless overridden all codecs supports both text and binary format
+        return EnumSet.of(FORMAT_TEXT, FORMAT_BINARY);
+    }
+
     /**
      * Create a {@link EncodedParameter}.
      *
@@ -214,19 +223,4 @@ boolean isTypeAssignable(Class<?> type) {
         return type.isAssignableFrom(this.type);
     }
 
-    static class SimplePostgresTypeIdentifier implements PostgresTypeIdentifier {
-
-        private final int oid;
-
-        public SimplePostgresTypeIdentifier(int oid) {
-            this.oid = oid;
-        }
-
-        @Override
-        public int getObjectId() {
-            return this.oid;
-        }
-
-    }
-
 }

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

@@ -24,6 +24,7 @@
 import io.r2dbc.postgresql.util.ByteBufUtils;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -112,6 +113,11 @@ public EncodedParameter encodeNull() {
         return createNull(Format.FORMAT_BINARY, this.postgresqlObjectId);
     }
 
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return Collections.singleton(this.postgresqlObjectId);
+    }
+
     /**
      * Create a {@link TokenStream} given {@code content}.
      *

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

@@ -20,12 +20,18 @@
 import io.r2dbc.postgresql.message.Format;
 import io.r2dbc.postgresql.util.Assert;
 
+import java.util.EnumSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.JSON;
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.JSONB;
 import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
 
 abstract class AbstractJsonCodec<T> extends AbstractCodec<T> {
 
+    private static final Set<PostgresqlObjectId> SUPPORTED_TYPES = EnumSet.of(JSON, JSONB);
+
     AbstractJsonCodec(Class<T> type) {
         super(type);
     }
@@ -45,7 +51,12 @@ boolean doCanDecode(PostgresqlObjectId type, Format format) {
         Assert.requireNonNull(format, "format must not be null");
         Assert.requireNonNull(type, "type must not be null");
 
-        return JSONB == type || JSON == type;
+        return SUPPORTED_TYPES.contains(type);
+    }
+
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return SUPPORTED_TYPES.stream().map(PostgresTypeIdentifier.class::cast).collect(Collectors.toList());
     }
 
 }

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

@@ -27,6 +27,7 @@
 import java.util.EnumSet;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.FLOAT4;
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.FLOAT8;
@@ -144,6 +145,11 @@ public EncodedParameter encodeNull() {
         return createNull(FORMAT_BINARY, getDefaultType());
     }
 
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return SUPPORTED_TYPES.stream().map(PostgresTypeIdentifier.class::cast).collect(Collectors.toList());
+    }
+
     /**
      * Returns the {@link PostgresqlObjectId} for to identify whether this codec is the default codec.
      *

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

@@ -26,6 +26,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -202,6 +203,11 @@ EncodedParameter doEncode(Object[] value, PostgresTypeIdentifier dataType) {
         }, dataType);
     }
 
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return Collections.singleton(this.dataType);
+    }
+
     /**
      * Create the encoded array representation.
      *

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

@@ -28,6 +28,7 @@
 import reactor.util.annotation.Nullable;
 
 import java.nio.ByteBuffer;
+import java.util.Collections;
 
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.BYTEA;
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
@@ -79,6 +80,11 @@ EncodedParameter doEncode(Blob value, PostgresTypeIdentifier dataType) {
         );
     }
 
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return Collections.singleton(BYTEA);
+    }
+
     private static final class ByteABlob implements Blob {
 
         private final ByteBuf byteBuf;

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

@@ -22,6 +22,7 @@
 import io.r2dbc.postgresql.util.Assert;
 import io.r2dbc.postgresql.util.ByteBufUtils;
 
+import java.util.Collections;
 import java.util.function.Function;
 
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
@@ -87,6 +88,11 @@ public final EncodedParameter encodeNull() {
         return createNull(FORMAT_TEXT, this.postgresType);
     }
 
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return Collections.singleton(this.postgresType);
+    }
+
     @Override
     public final PostgresTypeIdentifier getArrayDataType() {
         return this.postgresArrayType;

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

@@ -78,4 +78,14 @@ public PostgresTypeIdentifier getArrayDataType() {
         return PostgresqlObjectId.INT2_ARRAY;
     }
 
+    @Override
+    public Iterable<Format> getFormats() {
+        return this.delegate.getFormats();
+    }
+
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return this.delegate.getDataTypes();
+    }
+
 }

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

@@ -80,4 +80,14 @@ public PostgresTypeIdentifier getArrayDataType() {
         return PostgresqlObjectId.CHAR_ARRAY;
     }
 
+    @Override
+    public Iterable<Format> getFormats() {
+        return this.delegate.getFormats();
+    }
+
+    @Override
+    public Iterable<PostgresTypeIdentifier> getDataTypes() {
+        return this.delegate.getDataTypes();
+    }
+
 }