Implement basic BATCH in Firebird Proxy

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/FirebirdCommandPacketFactory.java

@@ -21,6 +21,10 @@
 import lombok.NoArgsConstructor;
 import org.apache.shardingsphere.database.protocol.firebird.constant.protocol.FirebirdProtocolVersion;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.admin.FirebirdUnsupportedCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchCancelCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchCreateCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchExecuteCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.batch.FirebirdBatchSendMessageCommandPacket;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.blob.FirebirdGetBlobSegmentCommandPacket;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.blob.FirebirdOpenBlobCommandPacket;
 import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.blob.FirebirdPutBlobSegmentCommandPacket;
@@ -97,6 +101,14 @@ public static FirebirdCommandPacket newInstance(final FirebirdCommandPacketType
                 return new FirebirdRollbackTransactionPacket(payload);
             case FREE_STATEMENT:
                 return new FirebirdFreeStatementPacket(payload);
+            case BATCH_CREATE:
+                return new FirebirdBatchCreateCommandPacket(payload);
+            case BATCH_MSG:
+                return new FirebirdBatchSendMessageCommandPacket(payload);
+            case BATCH_EXEC:
+                return new FirebirdBatchExecuteCommandPacket(payload);
+            case BATCH_CANCEL:
+                return new FirebirdBatchCancelCommandPacket(payload);
             default:
                 return new FirebirdUnsupportedCommandPacket(commandPacketType);
         }
@@ -154,6 +166,14 @@ private static int getLength(final FirebirdCommandPacketType commandPacketType,
                 return FirebirdRollbackTransactionPacket.getLength();
             case FREE_STATEMENT:
                 return FirebirdFreeStatementPacket.getLength();
+            case BATCH_CREATE:
+                return FirebirdBatchCreateCommandPacket.getLength(payload);
+            case BATCH_MSG:
+                return FirebirdBatchSendMessageCommandPacket.getLength(payload);
+            case BATCH_EXEC:
+                return FirebirdBatchExecuteCommandPacket.getLength();
+            case BATCH_CANCEL:
+                return FirebirdBatchCancelCommandPacket.getLength();
             default:
                 return 0;
         }

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchCancelCommandPacket.java

@@ -0,0 +1,49 @@
+/*
+ * 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.database.protocol.firebird.packet.command.query.batch;
+
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+/**
+ * Firebird batch cancel command packet.
+ */
+@Getter
+public final class FirebirdBatchCancelCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    public FirebirdBatchCancelCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @return length of packet
+     */
+    public static int getLength() {
+        return 8;
+    }
+}

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchCreateCommandPacket.java

@@ -0,0 +1,65 @@
+/*
+ * 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.database.protocol.firebird.packet.command.query.batch;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.statement.FirebirdBlrRowMetadata;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+/**
+ * Firebird create batch command packet.
+ */
+@Getter
+public final class FirebirdBatchCreateCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    private final FirebirdBlrRowMetadata batchBlr;
+
+    private final long batchMessageLength;
+
+    private final ByteBuf batchParametersBuffer;
+
+    public FirebirdBatchCreateCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+        batchBlr = FirebirdBlrRowMetadata.parseBLR(payload.readBuffer());
+        batchMessageLength = payload.readInt4Unsigned();
+        batchParametersBuffer = payload.readBuffer();
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @param payload Firebird packet payload
+     * @return length of packet
+     */
+    public static int getLength(final FirebirdPacketPayload payload) {
+        int length = 8;
+        length += payload.getBufferLength(length);
+        length += 4;
+        length += payload.getBufferLength(length);
+        return length;
+    }
+}

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchExecuteCommandPacket.java

@@ -0,0 +1,49 @@
+/*
+ * 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.database.protocol.firebird.packet.command.query.batch;
+
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+@Getter
+public class FirebirdBatchExecuteCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    private final int transactionHandle;
+
+    public FirebirdBatchExecuteCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+        transactionHandle = payload.readInt4();
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @return length of packet
+     */
+    public static int getLength() {
+        return 12;
+    }
+}

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/batch/FirebirdBatchSendMessageCommandPacket.java

@@ -0,0 +1,67 @@
+/*
+ * 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.database.protocol.firebird.packet.command.query.batch;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.FirebirdCommandPacket;
+import org.apache.shardingsphere.database.protocol.firebird.payload.FirebirdPacketPayload;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Firebird batch message command packet.
+ */
+@Getter
+public final class FirebirdBatchSendMessageCommandPacket extends FirebirdCommandPacket {
+
+    private final int statementHandle;
+
+    private final long batchMessageCount;
+
+    private final byte[] batchData;
+
+    private final List<Object> parameterValues = new ArrayList<>();
+
+    public FirebirdBatchSendMessageCommandPacket(final FirebirdPacketPayload payload) {
+        payload.skipReserved(4);
+        statementHandle = payload.readInt4();
+        batchMessageCount = payload.readInt4Unsigned();
+        ByteBuf buf = payload.getByteBuf();
+        int remaining = buf.readableBytes();
+        batchData = new byte[remaining];
+        buf.readBytes(batchData);
+    }
+
+    @Override
+    protected void write(final FirebirdPacketPayload payload) {
+    }
+
+    /**
+     * Get length of packet.
+     *
+     * @param payload Firebird packet payload
+     * @return length of packet
+     */
+    public static int getLength(final FirebirdPacketPayload payload) {
+        // TODO Do not rely on fixed header subtraction. Implement proper packet length calculation by parsing BATCH_MSG fields.
+        int readable = payload.getByteBuf().readableBytes();
+        return readable > 12 ? readable - 12 : -1;
+    }
+}

database/protocol/dialect/firebird/src/main/java/org/apache/shardingsphere/database/protocol/firebird/packet/command/query/statement/FirebirdBlrRowMetadata.java

@@ -0,0 +1,98 @@
+/*
+ * 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.database.protocol.firebird.packet.command.query.statement;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Getter;
+import org.apache.shardingsphere.database.protocol.firebird.packet.command.query.FirebirdBinaryColumnType;
+import org.firebirdsql.gds.BlrConstants;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Firebird BLR row metadata.
+ */
+@Getter
+public final class FirebirdBlrRowMetadata {
+
+    private final ByteBuf blr;
+
+    private final int length;
+
+    private final List<FirebirdBinaryColumnType> columnTypes;
+
+    private FirebirdBlrRowMetadata(final ByteBuf blr, final int length, final List<FirebirdBinaryColumnType> columnTypes) {
+        this.blr = blr;
+        this.length = length;
+        this.columnTypes = columnTypes;
+    }
+
+    /**
+     * Parse FirebirdBlrRowMetadata from BLR buffer.
+     *
+     * @param blrBuffer BLR buffer
+     * @return row binary
+     */
+    public static FirebirdBlrRowMetadata parseBLR(final ByteBuf blrBuffer) {
+        int length = blrBuffer.readableBytes();
+        List<FirebirdBinaryColumnType> columnTypes = parse(blrBuffer);
+        return new FirebirdBlrRowMetadata(blrBuffer, length, columnTypes);
+    }
+
+    private static List<FirebirdBinaryColumnType> parse(final ByteBuf blrBuffer) {
+        ByteBuf buffer = blrBuffer.duplicate();
+        if (!buffer.isReadable()) {
+            return new ArrayList<>(0);
+        }
+        buffer.skipBytes(4);
+        int length = buffer.readUnsignedByte();
+        length += 256 * buffer.readUnsignedByte();
+        List<FirebirdBinaryColumnType> result = new ArrayList<>(length / 2);
+        int blrType = buffer.readUnsignedByte();
+        while (blrType != BlrConstants.blr_end) {
+            FirebirdBinaryColumnType type = FirebirdBinaryColumnType.valueOfBLRType(blrType);
+            result.add(type);
+            buffer.skipBytes(getSkipCount(type) + 2);
+            blrType = buffer.readUnsignedByte();
+        }
+        return result;
+    }
+
+    private static int getSkipCount(final FirebirdBinaryColumnType type) {
+        switch (type) {
+            case VARYING:
+            case TEXT:
+                return 4;
+            case NULL:
+            case LEGACY_TEXT:
+            case LEGACY_VARYING:
+                return 2;
+            case BLOB:
+            case ARRAY:
+            case LONG:
+            case SHORT:
+            case INT64:
+            case QUAD:
+            case INT128:
+                return 1;
+            default:
+                return 0;
+        }
+    }
+}