Parameter values are not sent in the correct order

[alightgoesout]

Bug Report

I have encountered a bug when using custom types and codecs, where parameters in the bind command are not set in the right order. I have created a minimal repository with more details, including frames captured by Wireshark.

Versions

• Driver: 0.9.0.RELEASE
• Database: PostgreSQL 13.5 (Debian 13.5-1.pgdg110+1)
• Java: 11.0.13 (Red Hat, Inc. 11.0.13+8)
• OS: Fedora 35

Current Behavior

I have a table with three custom types (a composite type, an enum, and a range). When I try to insert a row using createStatement, the parameters value are not sent in the correct order. The parse command is in the right order with the correct OIDs, but the bind command permutes some parameters and the command is thus rejected by the database.

Stack trace

Exception in thread "main" io.r2dbc.postgresql.ExceptionFactory$PostgresqlBadGrammarException: [22P02] malformed range literal: "(one,2,3.3)"
	at io.r2dbc.postgresql.ExceptionFactory.createException(ExceptionFactory.java:96)
	at io.r2dbc.postgresql.ExceptionFactory.createException(ExceptionFactory.java:65)
	at io.r2dbc.postgresql.ExceptionFactory.handleErrorResponse(ExceptionFactory.java:132)
	at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:169)
	at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
	at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
	at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
	at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
	at io.r2dbc.postgresql.util.FluxDiscardOnCancel$FluxDiscardOnCancelSubscriber.onNext(FluxDiscardOnCancel.java:91)
	at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onNext(FluxDoFinally.java:130)
	at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:119)
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:793)
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.next(FluxCreate.java:718)
	at reactor.core.publisher.FluxCreate$SerializedFluxSink.next(FluxCreate.java:154)
	at io.r2dbc.postgresql.client.ReactorNettyClient$Conversation.emit(ReactorNettyClient.java:617)
	at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.emit(ReactorNettyClient.java:868)
	at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.onNext(ReactorNettyClient.java:742)
	at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.onNext(ReactorNettyClient.java:649)
	at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:119)
	at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
	at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:220)
	at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:220)
	at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:279)
	at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:388)
	at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:404)
	at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:795)
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:480)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:829)

Table schema

Input Code

CREATE TYPE item_tuple AS (one text, two integer, three double precision);

CREATE TYPE item_status AS ENUM ('GOOD', 'BAD');

CREATE TYPE item_range AS RANGE (SUBTYPE = integer);

CREATE TABLE items (
  id uuid primary key,
  name text not null,
  tuple item_tuple not null,
  status item_status not null,
  range item_range not null
);

Steps to reproduce

I have created a minimal repository to reproduce the bug. I have created custom codecs for the custom types. I just create an insert statement using createStatement, then binds the correct values and execute the statement.

Expected behavior/code

The statement should succeed and a new row should be inserted in the database.

[mp911de]

Thanks for report. For some reason, the issue only becomes apparent when using Coroutines or a non-main thread. Parameter value collection doesn't consider the actual ordering and therefore the actual values are collected in random order.