Fix #6579 : Support PostgreSQL unspecified parameter data type (#6998)

* Support unspecified parameter data type (#6579)

* Support unspecified parameter data type (#6579)

* Support PostgreSQL unspecified parameter data type

* Support unspecified parameter data type, follow review suggestion

shardingsphere-db-protocol/shardingsphere-db-protocol-core/src/main/java/org/apache/shardingsphere/db/protocol/parameter/TypeUnspecifiedSQLParameter.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *     http://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 org.apache.shardingsphere.db.protocol.parameter;
+
+/**
+ * Type unspecified SQL parameter.
+ */
+public interface TypeUnspecifiedSQLParameter {
+}

shardingsphere-db-protocol/shardingsphere-db-protocol-postgresql/src/main/java/org/apache/shardingsphere/db/protocol/postgresql/packet/command/query/binary/bind/PostgreSQLComBindPacket.java

@@ -70,7 +70,11 @@ public final class PostgreSQLComBindPacket extends PostgreSQLCommandPacket {
         int parametersCount = payload.readInt2();
         List<Object> result = new ArrayList<>(parametersCount);
         for (int parameterIndex = 0; parameterIndex < parametersCount; parameterIndex++) {
-            payload.readInt4();
+            int paramValueLen = payload.readInt4();
+            if (-1 == paramValueLen) {
+                result.add(null);
+                continue;
+            }
             PostgreSQLBinaryProtocolValue binaryProtocolValue = PostgreSQLBinaryProtocolValueFactory.getBinaryProtocolValue(parameterTypes.get(parameterIndex).getColumnType());
             result.add(binaryProtocolValue.read(payload));
         }

shardingsphere-db-protocol/shardingsphere-db-protocol-postgresql/src/main/java/org/apache/shardingsphere/db/protocol/postgresql/packet/command/query/binary/bind/PostgreSQLTypeUnspecifiedSQLParameter.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *     http://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 org.apache.shardingsphere.db.protocol.postgresql.packet.command.query.binary.bind;
+
+import org.apache.shardingsphere.db.protocol.parameter.TypeUnspecifiedSQLParameter;
+
+/**
+ * Type unspecified SQL parameter for PostgreSQL.
+ */
+public final class PostgreSQLTypeUnspecifiedSQLParameter implements TypeUnspecifiedSQLParameter {
+    
+    private final String parameterValue;
+    
+    public PostgreSQLTypeUnspecifiedSQLParameter(final String parameterValue) {
+        this.parameterValue = parameterValue;
+    }
+    
+    @Override
+    public String toString() {
+        return parameterValue;
+    }
+}

shardingsphere-db-protocol/shardingsphere-db-protocol-postgresql/src/main/java/org/apache/shardingsphere/db/protocol/postgresql/packet/command/query/binary/bind/protocol/PostgreSQLBinaryProtocolValueFactory.java

@@ -34,6 +34,7 @@ public final class PostgreSQLBinaryProtocolValueFactory {
     private static final Map<PostgreSQLColumnType, PostgreSQLBinaryProtocolValue> BINARY_PROTOCOL_VALUES = new HashMap<>();
     
     static {
+        setUnspecifiedBinaryProtocolValue();
         setStringLenencBinaryProtocolValue();
         setInt8BinaryProtocolValue();
         setInt4BinaryProtocolValue();
@@ -44,6 +45,11 @@ public final class PostgreSQLBinaryProtocolValueFactory {
         setTimeBinaryProtocolValue();
     }
     
+    private static void setUnspecifiedBinaryProtocolValue() {
+        PostgreSQLUnspecifiedBinaryProtocolValue binaryProtocolValue = new PostgreSQLUnspecifiedBinaryProtocolValue();
+        BINARY_PROTOCOL_VALUES.put(PostgreSQLColumnType.POSTGRESQL_TYPE_UNSPECIFIED, binaryProtocolValue);
+    }
+    
     private static void setStringLenencBinaryProtocolValue() {
         PostgreSQLStringBinaryProtocolValue binaryProtocolValue = new PostgreSQLStringBinaryProtocolValue();
         BINARY_PROTOCOL_VALUES.put(PostgreSQLColumnType.POSTGRESQL_TYPE_VARCHAR, binaryProtocolValue);

shardingsphere-db-protocol/shardingsphere-db-protocol-postgresql/src/main/java/org/apache/shardingsphere/db/protocol/postgresql/packet/command/query/binary/bind/protocol/PostgreSQLUnspecifiedBinaryProtocolValue.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *     http://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 org.apache.shardingsphere.db.protocol.postgresql.packet.command.query.binary.bind.protocol;
+
+import org.apache.shardingsphere.db.protocol.postgresql.packet.command.query.binary.bind.PostgreSQLTypeUnspecifiedSQLParameter;
+import org.apache.shardingsphere.db.protocol.postgresql.payload.PostgreSQLPacketPayload;
+
+/**
+ * Binary protocol value for unspecified for PostgreSQL.
+ */
+public final class PostgreSQLUnspecifiedBinaryProtocolValue implements PostgreSQLBinaryProtocolValue {
+    
+    @Override
+    public int getColumnLength(final Object value) {
+        throw new UnsupportedOperationException();
+    }
+    
+    @Override
+    public Object read(final PostgreSQLPacketPayload payload) {
+        payload.getByteBuf().readerIndex(payload.getByteBuf().readerIndex() - 4);
+        byte[] bytes = new byte[payload.readInt4()];
+        payload.getByteBuf().readBytes(bytes);
+        String result = new String(bytes);
+        return new PostgreSQLTypeUnspecifiedSQLParameter(result);
+    }
+    
+    @Override
+    public void write(final PostgreSQLPacketPayload payload, final Object value) {
+        throw new UnsupportedOperationException();
+    }
+}

shardingsphere-db-protocol/shardingsphere-db-protocol-postgresql/src/test/java/org/apache/shardingsphere/db/protocol/postgresql/packet/command/query/binary/bind/PostgreSQLTypeUnspecifiedSQLParameterTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *     http://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 org.apache.shardingsphere.db.protocol.postgresql.packet.command.query.binary.bind;
+
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public final class PostgreSQLTypeUnspecifiedSQLParameterTest {
+    
+    @Test
+    public void assertToString() {
+        String timestampStr = "2020-08-23 15:57:03+08";
+        PostgreSQLTypeUnspecifiedSQLParameter parameter = new PostgreSQLTypeUnspecifiedSQLParameter(timestampStr);
+        assertThat(parameter.toString(), is(timestampStr));
+    }
+    
+}

shardingsphere-db-protocol/shardingsphere-db-protocol-postgresql/src/test/java/org/apache/shardingsphere/db/protocol/postgresql/packet/command/query/binary/bind/protocol/PostgreSQLUnspecifiedBinaryProtocolValueTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *     http://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 org.apache.shardingsphere.db.protocol.postgresql.packet.command.query.binary.bind.protocol;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.UnpooledHeapByteBuf;
+import java.nio.charset.StandardCharsets;
+import org.apache.shardingsphere.db.protocol.postgresql.packet.command.query.binary.bind.PostgreSQLTypeUnspecifiedSQLParameter;
+import org.apache.shardingsphere.db.protocol.postgresql.payload.PostgreSQLPacketPayload;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+public final class PostgreSQLUnspecifiedBinaryProtocolValueTest {
+    
+    @Test(expected = UnsupportedOperationException.class)
+    public void assertGetColumnLength() {
+        new PostgreSQLUnspecifiedBinaryProtocolValue().getColumnLength("val");
+    }
+    
+    @Test
+    public void assertRead() {
+        String timestampStr = "2020-08-23 15:57:03+08";
+        PooledByteBufAllocator byteBufAllocator = new PooledByteBufAllocator();
+        ByteBuf byteBuf = new UnpooledHeapByteBuf(byteBufAllocator, 4 + timestampStr.length(), 64);
+        byteBuf.writeInt(timestampStr.length());
+        byteBuf.writeCharSequence(timestampStr, StandardCharsets.ISO_8859_1);
+        byteBuf.readInt();
+        PostgreSQLPacketPayload payload = new PostgreSQLPacketPayload(byteBuf);
+        Object result = new PostgreSQLUnspecifiedBinaryProtocolValue().read(payload);
+        byteBuf.release();
+        assertNotNull(result);
+        assertTrue(result instanceof PostgreSQLTypeUnspecifiedSQLParameter);
+        assertThat(result.toString(), is(timestampStr));
+    }
+    
+    @Test(expected = UnsupportedOperationException.class)
+    public void assertWrite() {
+        new PostgreSQLUnspecifiedBinaryProtocolValue().write(mock(PostgreSQLPacketPayload.class), "val");
+    }
+    
+}

shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/jdbc/connection/BackendConnection.java

@@ -20,10 +20,12 @@ package org.apache.shardingsphere.proxy.backend.communication.jdbc.connection;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.LinkedHashMultimap;
 import com.google.common.collect.Multimap;
+import java.sql.Types;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.db.protocol.parameter.TypeUnspecifiedSQLParameter;
 import org.apache.shardingsphere.infra.database.type.DatabaseType;
 import org.apache.shardingsphere.infra.database.type.dialect.MySQLDatabaseType;
 import org.apache.shardingsphere.infra.database.type.dialect.PostgreSQLDatabaseType;
@@ -204,7 +206,12 @@ public final class BackendConnection implements JDBCExecutionConnection, AutoClo
         PreparedStatement result = option.isReturnGeneratedKeys()
                 ? connection.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS) : connection.prepareStatement(sql);
         for (int i = 0; i < parameters.size(); i++) {
-            result.setObject(i + 1, parameters.get(i));
+            Object parameter = parameters.get(i);
+            if (parameter instanceof TypeUnspecifiedSQLParameter) {
+                result.setObject(i + 1, parameter, Types.OTHER);
+            } else {
+                result.setObject(i + 1, parameter);
+            }
         }
         if (ConnectionMode.MEMORY_STRICTLY == connectionMode) {
             setFetchSize(result);

shardingsphere-proxy/shardingsphere-proxy-frontend/shardingsphere-proxy-frontend-postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/PostgreSQLErrPacketFactory.java

@@ -39,10 +39,16 @@ public final class PostgreSQLErrPacketFactory {
         if (cause instanceof PSQLException) {
             ServerErrorMessage serverErrorMessage = ((PSQLException) cause).getServerErrorMessage();
             PostgreSQLErrorResponsePacket errorResponsePacket = new PostgreSQLErrorResponsePacket();
-            errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_SEVERITY, serverErrorMessage.getSeverity());
-            errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_CODE, serverErrorMessage.getSQLState());
-            errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_MESSAGE, serverErrorMessage.getMessage());
-            errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_POSITION, Integer.toString(serverErrorMessage.getPosition()));
+            if (null != serverErrorMessage) {
+                errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_SEVERITY, serverErrorMessage.getSeverity());
+                errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_CODE, serverErrorMessage.getSQLState());
+                errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_MESSAGE, serverErrorMessage.getMessage());
+                errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_POSITION, Integer.toString(serverErrorMessage.getPosition()));
+            } else {
+                PSQLException ex = (PSQLException) cause;
+                errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_CODE, ex.getSQLState());
+                errorResponsePacket.addField(PostgreSQLErrorResponsePacket.FIELD_TYPE_MESSAGE, ex.getMessage());
+            }
             return errorResponsePacket;
         }
         PostgreSQLErrorResponsePacket errorResponsePacket = new PostgreSQLErrorResponsePacket();