diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlCursorFetchCompatibility.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlCursorFetchCompatibility.java new file mode 100644 index 00000000000000..981ce75da470da --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlCursorFetchCompatibility.java @@ -0,0 +1,61 @@ +// 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.doris.mysql; + +import com.google.common.collect.ImmutableSet; + +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +/** Resolves the incompatible cursor result-set behavior used by Connector/J releases. */ +public final class MysqlCursorFetchCompatibility { + private static final Set MYSQL_CONNECTOR_J_CLIENT_NAMES = ImmutableSet.of( + "MySQL Connector/J", "MySQL Connector Java"); + private static final Pattern CONSUMES_METADATA_TERMINATOR = + Pattern.compile("^(?:(?:5|6|8)\\.|9\\.[0-4](?:\\.|$))"); + private static final Pattern VERSION = Pattern.compile("^\\d+(?:\\.\\d+)+(?:[-+].*)?$"); + + public enum Behavior { + CONSUMES_METADATA_TERMINATOR, + STANDARD, + UNKNOWN + } + + private MysqlCursorFetchCompatibility() { + } + + public static Behavior resolve(Map connectAttributes) { + String clientName = connectAttributes.get("_client_name"); + if (clientName == null) { + return Behavior.UNKNOWN; + } + if (!MYSQL_CONNECTOR_J_CLIENT_NAMES.contains(clientName)) { + return Behavior.STANDARD; + } + + String clientVersion = connectAttributes.get("_client_version"); + if (clientVersion == null || !VERSION.matcher(clientVersion).matches()) { + return Behavior.UNKNOWN; + } + if (CONSUMES_METADATA_TERMINATOR.matcher(clientVersion).find()) { + return Behavior.CONSUMES_METADATA_TERMINATOR; + } + return Behavior.STANDARD; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlOkPacket.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlOkPacket.java index 4fa80102317417..f9cf43b36919fd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlOkPacket.java +++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlOkPacket.java @@ -58,16 +58,11 @@ public void writeTo(MysqlSerializer serializer) { // TODO(zhaochun): STATUS_FLAGS // if ((STATUS_FLAGS & MysqlStatusFlag.SERVER_SESSION_STATE_CHANGED) != 0) { // } - } else { - // Always write the info field as a length-encoded string. - // When CLIENT_DEPRECATE_EOF is negotiated, the driver's OkPacket.parse() - // unconditionally reads STRING_LENENC for info, so an empty string must - // still be written (as a single 0x00 byte representing length 0). - if (Strings.isNullOrEmpty(infoMessage)) { - serializer.writeVInt(0); - } else { - serializer.writeLenEncodedString(infoMessage); - } + } else if (!Strings.isNullOrEmpty(infoMessage)) { + serializer.writeLenEncodedString(infoMessage); + } else if (capability.isDeprecatedEOF()) { + // Connector/J parses the info field for CLIENT_DEPRECATE_EOF even when it is empty. + serializer.writeVInt(0); } } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlResultSetEndPacket.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlResultSetEndPacket.java index 5543b85c361d8b..fd76924949daab 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlResultSetEndPacket.java +++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlResultSetEndPacket.java @@ -40,6 +40,7 @@ public class MysqlResultSetEndPacket extends MysqlPacket { public MysqlResultSetEndPacket(QueryState state) { this.serverStatus = state.serverStatus; + this.warningCount = state.getWarningRows(); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java index 428f86c957e92d..43c8edb6e26abd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java @@ -283,6 +283,9 @@ public enum ConnectType { @Setter private ByteBuffer prepareExecuteBuffer; + // Whether the current COM_STMT_EXECUTE requested a server-side read-only cursor. + private boolean cursorFetchRequested; + private MysqlHandshakePacket mysqlHandshakePacket; public void setUserQueryTimeout(int queryTimeout) { @@ -509,6 +512,14 @@ public void setConnectAttributes(Map connectAttributes) { this.connectAttributes = new HashMap<>(connectAttributes); } + public boolean isCursorFetchRequested() { + return cursorFetchRequested; + } + + public void setCursorFetchRequested(boolean cursorFetchRequested) { + this.cursorFetchRequested = cursorFetchRequested; + } + public boolean isTxnModel() { return txnEntry != null && txnEntry.isTxnModel(); } @@ -997,6 +1008,7 @@ public void clear() { statementContext = null; loadBackendSelectionDecision = null; loadBackendSelectionHint = null; + cursorFetchRequested = false; } // Arrow Flight SQL only. diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java index 4fe8515421003c..97a7b7bbb4d5f2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java @@ -654,8 +654,23 @@ public void finalizeCommand() throws IOException { && ctx.getState().getStateType() != QueryState.MysqlStateType.ERR) { ShowResultSet resultSet = executor.getShowResultSet(); if (resultSet == null) { - executor.sendProxyQueryResult(); - packet = executor.getOutputPacket(); + if (ctx.getMysqlChannel().clientDeprecatedEOF() + && !executor.isForwardedClientDeprecatedEofApplied() + && executor.getProxyStatusCode() == 0) { + if (executor.hasForwardedQueryResultPackets()) { + ctx.getState().setError(ErrorCode.ERR_NOT_SUPPORTED_YET, + "The master FE cannot preserve CLIENT_DEPRECATE_EOF while forwarding this query. " + + "Connect to the master FE or finish the FE rolling upgrade"); + } else { + // An old master has already completed a DDL/DML operation. Rebuild its final OK locally + // instead of returning an upgrade error that could make the client retry side effects. + ctx.getState().setOk(executor.getForwardedAffectedRows(), 0, null); + } + packet = getResultPacket(); + } else { + executor.sendProxyQueryResult(); + packet = executor.getOutputPacket(); + } } else { executor.sendResultSet(resultSet); packet = getResultPacket(); @@ -729,6 +744,8 @@ public TMasterOpResult proxyExecute(TMasterOpRequest request) throws TException if (request.isSetClientDeprecatedEOF() && request.isClientDeprecatedEOF()) { ctx.getMysqlChannel().setClientDeprecatedEOF(); } + ctx.setCursorFetchRequested(request.isSetCursorFetchRequested() + && request.isCursorFetchRequested()); ctx.setThreadLocalInfo(); StmtExecutor executor = null; @@ -818,6 +835,7 @@ public TMasterOpResult proxyExecute(TMasterOpRequest request) throws TException ctx.getState().serverStatus |= MysqlServerStatusFlag.SERVER_MORE_RESULTS_EXISTS; } result.setPacket(getResultPacket()); + result.setClientDeprecatedEofApplied(ctx.getMysqlChannel().clientDeprecatedEOF()); result.setStatus(ctx.getState().toString()); if (ctx.getState().getStateType() == MysqlStateType.OK) { result.setStatusCode(0); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java index cb1f6d5da9e5c4..da521d81b6d67c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java @@ -214,6 +214,7 @@ protected TMasterOpRequest buildStmtForwardParams() throws AnalysisException { if (null != ctx.getPrepareExecuteBuffer()) { params.setPrepareExecuteBuffer(ctx.getPrepareExecuteBuffer()); } + params.setCursorFetchRequested(ctx.isCursorFetchRequested()); } ctx.getSessionContext().getDelegatedCredential().ifPresent((DelegatedCredential credential) -> { @@ -258,6 +259,20 @@ public ByteBuffer getOutputPacket() { return result.packet; } + public boolean isClientDeprecatedEofApplied() { + return result != null && result.isSetClientDeprecatedEofApplied() + && result.isClientDeprecatedEofApplied(); + } + + public boolean hasQueryResultPackets() { + return result != null && result.isSetQueryResultBufList() + && !result.getQueryResultBufList().isEmpty(); + } + + public long getAffectedRows() { + return result != null && result.isSetAffectedRows() ? result.getAffectedRows() : 0; + } + public TUniqueId getQueryId() { if (result != null && result.isSetQueryId()) { return result.getQueryId(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java index f70a9d3f622116..9e0d786591e0b9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java @@ -29,6 +29,7 @@ import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.mysql.MysqlChannel; import org.apache.doris.mysql.MysqlCommand; +import org.apache.doris.mysql.MysqlCursorFetchCompatibility; import org.apache.doris.mysql.MysqlHandshakePacket; import org.apache.doris.mysql.MysqlProto; import org.apache.doris.mysql.MysqlSerializer; @@ -62,6 +63,7 @@ */ public class MysqlConnectProcessor extends ConnectProcessor { private static final Logger LOG = LogManager.getLogger(MysqlConnectProcessor.class); + private static final int CURSOR_TYPE_READ_ONLY = 0x01; private ByteBuffer packetBuf; @@ -207,10 +209,18 @@ private void handleExecute() { packetBuf = packetBuf.order(ByteOrder.LITTLE_ENDIAN); // parse stmt_id, flags, params int stmtId = packetBuf.getInt(); - // flag - packetBuf.get(); + int flags = Byte.toUnsignedInt(packetBuf.get()); + ctx.setCursorFetchRequested((flags & CURSOR_TYPE_READ_ONLY) != 0); // iteration_count always 1, packetBuf.getInt(); + if (ctx.isCursorFetchRequested() && ctx.getMysqlChannel().clientDeprecatedEOF() + && MysqlCursorFetchCompatibility.resolve(ctx.getConnectAttributes()) + == MysqlCursorFetchCompatibility.Behavior.UNKNOWN) { + ctx.getState().setError(ErrorCode.ERR_NOT_SUPPORTED_YET, + "Cannot safely execute cursor fetch because the client did not provide identifiable " + + "connection attributes. Enable connection attributes or set useCursorFetch=false"); + return; + } if (LOG.isDebugEnabled()) { LOG.debug("execute prepared statement {}", stmtId); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java index 113985e1cd198b..cd0a3093c22a86 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java @@ -74,7 +74,9 @@ import org.apache.doris.mysql.FieldInfo; import org.apache.doris.mysql.MysqlChannel; import org.apache.doris.mysql.MysqlCommand; +import org.apache.doris.mysql.MysqlCursorFetchCompatibility; import org.apache.doris.mysql.MysqlEofPacket; +import org.apache.doris.mysql.MysqlResultSetEndPacket; import org.apache.doris.mysql.MysqlSerializer; import org.apache.doris.mysql.ProxyMysqlChannel; import org.apache.doris.nereids.NereidsPlanner; @@ -504,6 +506,18 @@ public ByteBuffer getOutputPacket() { } } + public boolean isForwardedClientDeprecatedEofApplied() { + return masterOpExecutor != null && masterOpExecutor.isClientDeprecatedEofApplied(); + } + + public boolean hasForwardedQueryResultPackets() { + return masterOpExecutor != null && masterOpExecutor.hasQueryResultPackets(); + } + + public long getForwardedAffectedRows() { + return masterOpExecutor == null ? 0 : masterOpExecutor.getAffectedRows(); + } + /** * Whether this executor has actually forwarded to master and created a {@link MasterOpExecutor}. * @@ -1843,15 +1857,7 @@ private void sendMetaData(ResultSetMetaData metaData, List fieldInfos } context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer()); } - // When CLIENT_DEPRECATE_EOF is set, the server should not send the intermediate - // EOF packet after column definitions. The client will go directly from column - // definitions to reading data rows. - if (!context.getMysqlChannel().clientDeprecatedEOF()) { - serializer.reset(); - MysqlEofPacket eofPacket = new MysqlEofPacket(context.getState()); - eofPacket.writeTo(serializer); - context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer()); - } + sendMetadataTerminatorIfNeeded(context.getMysqlChannel()); } private List exprToStringType(List exprs) { @@ -1987,17 +1993,30 @@ private void sendFields(List colNames, List fieldInfos, List< channel.sendOnePacket(serializer.toByteBuffer()); } } - // When CLIENT_DEPRECATE_EOF is set, the server should not send the intermediate - // EOF packet after column definitions. The client will go directly from column - // definitions to reading data rows. + sendMetadataTerminatorIfNeeded(channel); + } + + private void sendMetadataTerminatorIfNeeded(MysqlChannel channel) throws IOException { if (!channel.clientDeprecatedEOF()) { serializer.reset(); - MysqlEofPacket eofPacket = new MysqlEofPacket(context.getState()); - eofPacket.writeTo(serializer); + new MysqlEofPacket(context.getState()).writeTo(serializer); + channel.sendOnePacket(serializer.toByteBuffer()); + } else if (connectorJConsumesCursorMetadataTerminator()) { + // Connector/J before 9.5 consumes the first OK packet after column definitions + // while probing whether a requested cursor was created. Doris does not create a + // cursor, so an empty result would otherwise lose its only end marker and block. + serializer.reset(); + new MysqlResultSetEndPacket(context.getState()).writeTo(serializer); channel.sendOnePacket(serializer.toByteBuffer()); } } + private boolean connectorJConsumesCursorMetadataTerminator() { + return context.isCursorFetchRequested() + && MysqlCursorFetchCompatibility.resolve(context.getConnectAttributes()) + == MysqlCursorFetchCompatibility.Behavior.CONSUMES_METADATA_TERMINATOR; + } + public void sendResultSet(ResultSet resultSet) throws IOException { sendResultSet(resultSet, null); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlCursorFetchCompatibilityTest.java b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlCursorFetchCompatibilityTest.java new file mode 100644 index 00000000000000..4747725e128d0e --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlCursorFetchCompatibilityTest.java @@ -0,0 +1,59 @@ +// 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.doris.mysql; + +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; + +public class MysqlCursorFetchCompatibilityTest { + @Test + public void testConnectorJBehaviorBoundaries() { + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.CONSUMES_METADATA_TERMINATOR, + resolve("MySQL Connector Java", "5.1.49")); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.CONSUMES_METADATA_TERMINATOR, + resolve("MySQL Connector/J", "6.0.6")); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.CONSUMES_METADATA_TERMINATOR, + resolve("MySQL Connector/J", "8.2.0")); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.CONSUMES_METADATA_TERMINATOR, + resolve("MySQL Connector/J", "9.4.0")); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.STANDARD, + resolve("MySQL Connector/J", "9.5.0")); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.STANDARD, + resolve("MySQL Connector/J", "9.6.0")); + } + + @Test + public void testUnknownAndOtherClients() { + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.UNKNOWN, + MysqlCursorFetchCompatibility.resolve(Collections.emptyMap())); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.UNKNOWN, + MysqlCursorFetchCompatibility.resolve(ImmutableMap.of("_client_name", "MySQL Connector/J"))); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.UNKNOWN, + resolve("MySQL Connector/J", "custom")); + Assert.assertEquals(MysqlCursorFetchCompatibility.Behavior.STANDARD, + resolve("MariaDB Connector/J", "3.5.6")); + } + + private MysqlCursorFetchCompatibility.Behavior resolve(String clientName, String clientVersion) { + return MysqlCursorFetchCompatibility.resolve(ImmutableMap.of( + "_client_name", clientName, "_client_version", clientVersion)); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlOkPacketTest.java b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlOkPacketTest.java index 9fe47adbf5ebd1..7ac371dac4e536 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlOkPacketTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlOkPacketTest.java @@ -56,27 +56,20 @@ public void testWrite() { // assert warnings, int2: 0 Assert.assertEquals(0x00, MysqlProto.readInt2(buffer)); - // When infoMessage is empty, an empty len-encoded string (0x00) should still be written. - // This is required because OkPacket.parse() in MySQL Connector/J unconditionally reads - // STRING_LENENC for info. Without this byte, the driver throws - // ArrayIndexOutOfBoundsException when CLIENT_DEPRECATE_EOF is negotiated. - Assert.assertEquals(0x00, MysqlProto.readVInt(buffer)); Assert.assertEquals(0, buffer.remaining()); } @Test - public void testWritePayloadSizeGreaterThan5() { - // When CLIENT_DEPRECATE_EOF is negotiated, the driver distinguishes between - // EOF packets (payload <= 5) and ResultSet OK packets (payload > 5). - // MysqlOkPacket payload must be > 5 to avoid being misidentified as EOF. - // Payload: 0x00(1) + affected_rows(1) + last_insert_id(1) + status(2) + warnings(2) + info_len(1) = 8 + public void testWriteEmptyInfoWithDeprecatedEof() { + capability = new MysqlCapability(MysqlCapability.Flag.CLIENT_PROTOCOL_41.getFlagBit() + | MysqlCapability.Flag.CLIENT_DEPRECATE_EOF.getFlagBit()); MysqlOkPacket packet = new MysqlOkPacket(new QueryState()); MysqlSerializer serializer = MysqlSerializer.newInstance(capability); packet.writeTo(serializer); ByteBuffer buffer = serializer.toByteBuffer(); - int payloadLength = buffer.remaining(); - Assert.assertTrue("OK packet payload should be > 5 for CLIENT_DEPRECATE_EOF compatibility, got: " - + payloadLength, payloadLength > 5); + buffer.position(7); + Assert.assertEquals(0x00, MysqlProto.readVInt(buffer)); + Assert.assertEquals(0, buffer.remaining()); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlResultSetEndPacketTest.java b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlResultSetEndPacketTest.java index 8dc0d18877fde4..233811c93b42ba 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlResultSetEndPacketTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlResultSetEndPacketTest.java @@ -117,4 +117,30 @@ public void testDiffersFromEofPacket() { Assert.assertTrue("ResultSet OK packet payload should be > 5, got: " + rsEndPayloadLength, rsEndPayloadLength > 5); } + + @Test + public void testPreservesMoreResultsStatus() { + QueryState state = new QueryState(); + state.serverStatus = MysqlServerStatusFlag.SERVER_MORE_RESULTS_EXISTS; + MysqlSerializer serializer = MysqlSerializer.newInstance(capability); + new MysqlResultSetEndPacket(state).writeTo(serializer); + + ByteBuffer buffer = serializer.toByteBuffer(); + Assert.assertEquals(0xFE, MysqlProto.readInt1(buffer)); + Assert.assertEquals(0, MysqlProto.readVInt(buffer)); + Assert.assertEquals(0, MysqlProto.readVInt(buffer)); + Assert.assertEquals(MysqlServerStatusFlag.SERVER_MORE_RESULTS_EXISTS, MysqlProto.readInt2(buffer)); + } + + @Test + public void testPreservesWarningCount() { + QueryState state = new QueryState(); + state.setOk(0, 3, null); + MysqlSerializer serializer = MysqlSerializer.newInstance(capability); + new MysqlResultSetEndPacket(state).writeTo(serializer); + + ByteBuffer buffer = serializer.toByteBuffer(); + buffer.position(5); + Assert.assertEquals(3, MysqlProto.readInt2(buffer)); + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorForwardProtocolTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorForwardProtocolTest.java new file mode 100644 index 00000000000000..6324f3d1f244a5 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorForwardProtocolTest.java @@ -0,0 +1,144 @@ +// 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.doris.qe; + +import org.apache.doris.mysql.DummyMysqlChannel; +import org.apache.doris.mysql.MysqlCapability; +import org.apache.doris.mysql.MysqlProto; +import org.apache.doris.mysql.MysqlSerializer; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.IOException; +import java.nio.ByteBuffer; + +public class ConnectProcessorForwardProtocolTest { + @Test + public void testOldMasterQueryFailsBeforeRawPacketsAreSent() throws Exception { + TestContext context = new TestContext(); + StmtExecutor executor = forwardedExecutor(); + Mockito.when(executor.hasForwardedQueryResultPackets()).thenReturn(true); + + new TestProcessor(context, executor).finalizeCommand(); + + Assert.assertEquals(QueryState.MysqlStateType.ERR, context.getState().getStateType()); + Assert.assertEquals(0xFF, MysqlProto.readInt1(context.channel.packet)); + Mockito.verify(executor, Mockito.never()).sendProxyQueryResult(); + } + + @Test + public void testOldMasterDmlRebuildsOkWithoutRetryRisk() throws Exception { + TestContext context = new TestContext(); + StmtExecutor executor = forwardedExecutor(); + Mockito.when(executor.getForwardedAffectedRows()).thenReturn(7L); + + new TestProcessor(context, executor).finalizeCommand(); + + Assert.assertEquals(QueryState.MysqlStateType.OK, context.getState().getStateType()); + Assert.assertEquals(0x00, MysqlProto.readInt1(context.channel.packet)); + Assert.assertEquals(7L, MysqlProto.readVInt(context.channel.packet)); + Mockito.verify(executor, Mockito.never()).sendProxyQueryResult(); + } + + @Test + public void testRemoteErrorsRemainUnchanged() throws Exception { + TestContext context = new TestContext(); + StmtExecutor executor = forwardedExecutor(); + Mockito.when(executor.getProxyStatusCode()).thenReturn(1064); + Mockito.when(executor.getOutputPacket()).thenReturn(ByteBuffer.wrap(new byte[] {(byte) 0xFF, 1})); + + new TestProcessor(context, executor).finalizeCommand(); + + Assert.assertEquals(0xFF, MysqlProto.readInt1(context.channel.packet)); + Mockito.verify(executor).sendProxyQueryResult(); + } + + @Test + public void testNewMasterPacketsRemainUnchanged() throws Exception { + TestContext context = new TestContext(); + StmtExecutor executor = forwardedExecutor(); + Mockito.when(executor.isForwardedClientDeprecatedEofApplied()).thenReturn(true); + + new TestProcessor(context, executor).finalizeCommand(); + + Mockito.verify(executor).sendProxyQueryResult(); + } + + @Test + public void testLegacyEofClientDoesNotRequireConfirmation() throws Exception { + TestContext context = new TestContext(false); + StmtExecutor executor = forwardedExecutor(); + + new TestProcessor(context, executor).finalizeCommand(); + + Mockito.verify(executor).sendProxyQueryResult(); + } + + private StmtExecutor forwardedExecutor() { + StmtExecutor executor = Mockito.mock(StmtExecutor.class); + Mockito.when(executor.hasForwardedToMaster()).thenReturn(true); + Mockito.when(executor.getProxyStatusCode()).thenReturn(0); + return executor; + } + + private static class TestProcessor extends MysqlConnectProcessor { + private TestProcessor(ConnectContext context, StmtExecutor executor) { + super(context); + this.executor = executor; + } + } + + private static class TestContext extends ConnectContext { + private final RecordingChannel channel; + + private TestContext() { + this(true); + } + + private TestContext(boolean clientDeprecatedEof) { + channel = new RecordingChannel(clientDeprecatedEof); + } + + @Override + public RecordingChannel getMysqlChannel() { + return channel; + } + } + + private static class RecordingChannel extends DummyMysqlChannel { + private ByteBuffer packet; + + private RecordingChannel(boolean clientDeprecatedEof) { + int flags = MysqlCapability.Flag.CLIENT_PROTOCOL_41.getFlagBit(); + if (clientDeprecatedEof) { + flags |= MysqlCapability.Flag.CLIENT_DEPRECATE_EOF.getFlagBit(); + } + serializer = MysqlSerializer.newInstance(new MysqlCapability(flags)); + if (clientDeprecatedEof) { + setClientDeprecatedEOF(); + } + } + + @Override + public void sendAndFlush(ByteBuffer packet) throws IOException { + this.packet = packet.duplicate(); + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorMysqlProtocolTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorMysqlProtocolTest.java new file mode 100644 index 00000000000000..53b5186ab59bae --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorMysqlProtocolTest.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.doris.qe; + +import org.apache.doris.analysis.UserIdentity; +import org.apache.doris.catalog.Env; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.mysql.MysqlCommand; +import org.apache.doris.system.SystemInfoService; +import org.apache.doris.thrift.TMasterOpRequest; +import org.apache.doris.thrift.TMasterOpResult; +import org.apache.doris.thrift.TNetworkAddress; + +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.nio.ByteBuffer; +import java.util.Collections; + +public class FEOpExecutorMysqlProtocolTest { + @Test + public void testForwardRequestCarriesMysqlProtocolContext() throws Exception { + Env env = Mockito.mock(Env.class); + Mockito.when(env.getSelfNode()).thenReturn(new SystemInfoService.HostInfo("127.0.0.1", 9010)); + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + ConnectContext context = createContext(); + context.getMysqlChannel().setClientDeprecatedEOF(); + context.setCommand(MysqlCommand.COM_STMT_EXECUTE); + context.setCursorFetchRequested(true); + context.setConnectAttributes(ImmutableMap.of( + "_client_name", "MySQL Connector/J", "_client_version", "8.2.0")); + + TMasterOpRequest request = new TestFEOpExecutor(context).build(); + + Assert.assertTrue(request.isClientDeprecatedEOF()); + Assert.assertTrue(request.isCursorFetchRequested()); + Assert.assertEquals("8.2.0", request.getConnectAttributes().get("_client_version")); + } + } + + @Test + public void testForwardResponseRequiresExplicitProtocolConfirmation() { + TestFEOpExecutor executor = new TestFEOpExecutor(createContext()); + executor.setResult(new TMasterOpResult()); + Assert.assertFalse(executor.isClientDeprecatedEofApplied()); + Assert.assertFalse(executor.hasQueryResultPackets()); + Assert.assertEquals(0L, executor.getAffectedRows()); + + TMasterOpResult confirmed = new TMasterOpResult(); + confirmed.setClientDeprecatedEofApplied(true); + confirmed.setQueryResultBufList(Collections.singletonList(ByteBuffer.wrap(new byte[] {1}))); + confirmed.setAffectedRows(7); + executor.setResult(confirmed); + Assert.assertTrue(executor.isClientDeprecatedEofApplied()); + Assert.assertTrue(executor.hasQueryResultPackets()); + Assert.assertEquals(7L, executor.getAffectedRows()); + } + + private ConnectContext createContext() { + ConnectContext context = new ConnectContext(); + context.setCurrentUserIdentity(UserIdentity.createAnalyzedUserIdentWithIp("alice", "%")); + context.setRemoteIP("127.0.0.1"); + return context; + } + + private static class TestFEOpExecutor extends FEOpExecutor { + private TestFEOpExecutor(ConnectContext context) { + super(new TNetworkAddress("127.0.0.1", 9010), new OriginStatement("select 1", 0), context, true); + } + + private TMasterOpRequest build() throws AnalysisException { + return buildStmtForwardParams(); + } + + private void setResult(TMasterOpResult result) { + this.result = result; + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java new file mode 100644 index 00000000000000..0fa496eba57721 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java @@ -0,0 +1,78 @@ +// 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.doris.qe; + +import org.apache.doris.mysql.MysqlCommand; + +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; + +public class MysqlConnectProcessorCursorFetchTest { + private static final int CURSOR_TYPE_READ_ONLY = 1; + + @Test + public void testUnidentifiedDeprecatedEofCursorFailsBeforeExecution() throws Exception { + ConnectContext context = execute(true, true, false); + Assert.assertTrue(context.getState().getErrorMessage().contains( + "Cannot safely execute cursor fetch because the client did not provide identifiable")); + } + + @Test + public void testCompatibilityGateOnlyAppliesToAmbiguousProtocol() throws Exception { + Assert.assertTrue(execute(false, true, false).getState().getErrorMessage().contains( + "Unknown prepared statement handler")); + Assert.assertTrue(execute(true, false, false).getState().getErrorMessage().contains( + "Unknown prepared statement handler")); + Assert.assertTrue(execute(true, true, true).getState().getErrorMessage().contains( + "Unknown prepared statement handler")); + } + + private ConnectContext execute(boolean cursorRequested, boolean clientDeprecatedEof, + boolean identifiedClient) throws Exception { + ConnectContext context = new ConnectContext(); + context.setCommand(MysqlCommand.COM_STMT_EXECUTE); + if (clientDeprecatedEof) { + context.getMysqlChannel().setClientDeprecatedEOF(); + } + if (identifiedClient) { + context.setConnectAttributes(ImmutableMap.of( + "_client_name", "MySQL Connector/J", "_client_version", "8.2.0")); + } + + ByteBuffer packet = ByteBuffer.allocate(9).order(ByteOrder.LITTLE_ENDIAN); + packet.putInt(7); + packet.put((byte) (cursorRequested ? CURSOR_TYPE_READ_ONLY : 0)); + packet.putInt(1); + packet.flip(); + + MysqlConnectProcessor processor = new MysqlConnectProcessor(context); + Field packetField = MysqlConnectProcessor.class.getDeclaredField("packetBuf"); + packetField.setAccessible(true); + packetField.set(processor, packet); + Method handleExecute = MysqlConnectProcessor.class.getDeclaredMethod("handleExecute"); + handleExecute.setAccessible(true); + handleExecute.invoke(processor); + return context; + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java index 89bcdfd43f8bd2..16699281c742e5 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java @@ -37,6 +37,7 @@ import org.apache.doris.thrift.TUniqueId; import org.apache.doris.utframe.TestWithFeService; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import org.junit.Assert; import org.junit.jupiter.api.Assertions; @@ -49,7 +50,10 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; public class StmtExecutorTest extends TestWithFeService { @@ -333,6 +337,100 @@ public Void answer(InvocationOnMock invocation) { executor.sendBinaryResultRow(resultSet); } + @Test + public void testCursorFetchMetadataTerminatorDependsOnConnectorJVersion() throws IOException { + List connector82Packets = sendEmptyResultSet(true, "MySQL Connector/J", "8.2.0"); + Assertions.assertEquals(3, connector82Packets.size()); + Assertions.assertEquals(0xFE, Byte.toUnsignedInt(connector82Packets.get(2)[0])); + Assertions.assertTrue(connector82Packets.get(2).length > 5); + + Assertions.assertEquals(3, sendEmptyResultSet(true, "MySQL Connector Java", "5.1.49").size()); + Assertions.assertEquals(3, sendEmptyResultSet(true, "MySQL Connector/J", "6.0.6").size()); + Assertions.assertEquals(3, sendEmptyResultSet(true, "MySQL Connector/J", "9.4.0").size()); + Assertions.assertEquals(2, sendEmptyResultSet(true, "MySQL Connector/J", "9.5.0").size()); + Assertions.assertEquals(2, sendEmptyResultSet(false, "MySQL Connector/J", "8.2.0").size()); + Assertions.assertEquals(2, sendEmptyResultSet(true, "MariaDB Connector/J", "3.5.6").size()); + Assertions.assertEquals(2, sendEmptyResultSet(true, Collections.emptyMap()).size()); + + List legacyEofPackets = sendEmptyResultSet(true, "MySQL Connector/J", "8.2.0", false); + Assertions.assertEquals(3, legacyEofPackets.size()); + Assertions.assertEquals(5, legacyEofPackets.get(2).length); + } + + @Test + public void testPrepareMetadataTerminatorsFollowNegotiatedCapability() throws IOException { + Assertions.assertEquals(3, sendPrepareMetadata(false).size()); + Assertions.assertEquals(2, sendPrepareMetadata(true).size()); + } + + private List sendPrepareMetadata(boolean clientDeprecatedEof) throws IOException { + ConnectContext mockCtx = Mockito.mock(ConnectContext.class); + MysqlChannel channel = Mockito.mock(MysqlChannel.class); + Mockito.when(mockCtx.getConnectType()).thenReturn(ConnectType.MYSQL); + Mockito.when(mockCtx.getMysqlChannel()).thenReturn(channel); + Mockito.when(mockCtx.getState()).thenReturn(new QueryState()); + Mockito.when(mockCtx.getSessionVariable()).thenReturn(new SessionVariable()); + Mockito.when(channel.clientDeprecatedEOF()).thenReturn(clientDeprecatedEof); + Mockito.when(channel.getSerializer()).thenReturn(MysqlSerializer.newInstance()); + + List packets = new ArrayList<>(); + Mockito.doAnswer(invocation -> { + ByteBuffer packet = invocation.getArgument(0); + byte[] copy = new byte[packet.remaining()]; + packet.duplicate().get(copy); + packets.add(copy); + return null; + }).when(channel).sendOnePacket(Mockito.any(ByteBuffer.class)); + + new StmtExecutor(mockCtx, new OriginStatement("", 0), true).sendStmtPrepareOK( + 1, Collections.singletonList("p"), Collections.emptyList()); + return packets; + } + + private List sendEmptyResultSet(boolean cursorFetchRequested, String clientName, + String clientVersion) throws IOException { + return sendEmptyResultSet(cursorFetchRequested, clientName, clientVersion, true); + } + + private List sendEmptyResultSet(boolean cursorFetchRequested, String clientName, + String clientVersion, boolean clientDeprecatedEof) throws IOException { + return sendEmptyResultSet(cursorFetchRequested, ImmutableMap.of( + "_client_name", clientName, "_client_version", clientVersion), clientDeprecatedEof); + } + + private List sendEmptyResultSet(boolean cursorFetchRequested, + Map connectAttributes) throws IOException { + return sendEmptyResultSet(cursorFetchRequested, connectAttributes, true); + } + + private List sendEmptyResultSet(boolean cursorFetchRequested, + Map connectAttributes, boolean clientDeprecatedEof) throws IOException { + ConnectContext mockCtx = Mockito.mock(ConnectContext.class); + MysqlChannel channel = Mockito.mock(MysqlChannel.class); + Mockito.when(mockCtx.getConnectType()).thenReturn(ConnectType.MYSQL); + Mockito.when(mockCtx.getMysqlChannel()).thenReturn(channel); + Mockito.when(mockCtx.getState()).thenReturn(new QueryState()); + Mockito.when(mockCtx.getSessionVariable()).thenReturn(VariableMgr.newSessionVariable()); + Mockito.when(mockCtx.isCursorFetchRequested()).thenReturn(cursorFetchRequested); + Mockito.when(mockCtx.getConnectAttributes()).thenReturn(connectAttributes); + Mockito.when(channel.clientDeprecatedEOF()).thenReturn(clientDeprecatedEof); + Mockito.when(channel.getSerializer()).thenReturn(MysqlSerializer.newInstance()); + + List packets = new ArrayList<>(); + Mockito.doAnswer(invocation -> { + ByteBuffer packet = invocation.getArgument(0); + byte[] copy = new byte[packet.remaining()]; + packet.duplicate().get(copy); + packets.add(copy); + return null; + }).when(channel).sendOnePacket(Mockito.any(ByteBuffer.class)); + + List columns = Collections.singletonList(new Column("c", PrimitiveType.INT)); + ResultSet resultSet = new CommonResultSet(new CommonResultSetMetaData(columns), Collections.emptyList()); + new StmtExecutor(mockCtx, new OriginStatement("", 0), true).sendResultSet(resultSet); + return packets; + } + @Test public void testSendBinaryBooleanResultRow() throws IOException { ConnectContext mockCtx = Mockito.mock(ConnectContext.class); diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index f574b5bba40c3c..50e04e7881f798 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -441,6 +441,8 @@ struct TMasterOpRequest { 1005: optional string delegated_credential_token 1006: optional i64 delegated_credential_expires_at_millis 1007: optional string delegated_credential_session_id + // Whether COM_STMT_EXECUTE requested CURSOR_TYPE_READ_ONLY. + 1008: optional bool cursor_fetch_requested } struct TColumnDefinition { @@ -474,6 +476,8 @@ struct TMasterOpResult { 11: optional i64 affectedRows; // Lets the forwarding FE wait for the final statistics of external write fragments. 12: optional list auditStatisticsBackendIds; + // Confirms that the executing FE serialized raw MySQL packets with CLIENT_DEPRECATE_EOF. + 13: optional bool clientDeprecatedEofApplied; } // Certificate-based authentication info forwarded from BE to FE diff --git a/regression-test/data/prepared_stmt_p0/cursor_fetch_empty_result.out b/regression-test/data/prepared_stmt_p0/cursor_fetch_empty_result.out new file mode 100644 index 00000000000000..c2e5264c57cbef --- /dev/null +++ b/regression-test/data/prepared_stmt_p0/cursor_fetch_empty_result.out @@ -0,0 +1,5 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !empty_result -- + +-- !non_empty_result -- +1 diff --git a/regression-test/suites/arrow_flight_sql_p0/test_ddl.groovy b/regression-test/suites/arrow_flight_sql_p0/test_ddl.groovy new file mode 100644 index 00000000000000..c9dfa13cb3c581 --- /dev/null +++ b/regression-test/suites/arrow_flight_sql_p0/test_ddl.groovy @@ -0,0 +1,25 @@ +// 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. + +suite("test_ddl", "arrow_flight_sql") { + sql "DROP DATABASE IF EXISTS test_arrow_flight_sql_ddl" + + context.getArrowFlightSqlConnection().createStatement().withCloseable { statement -> + statement.execute("CREATE DATABASE test_arrow_flight_sql_ddl") + statement.execute("DROP DATABASE test_arrow_flight_sql_ddl") + } +} diff --git a/regression-test/suites/prepared_stmt_p0/cursor_fetch_empty_result.groovy b/regression-test/suites/prepared_stmt_p0/cursor_fetch_empty_result.groovy new file mode 100644 index 00000000000000..411598e0c83552 --- /dev/null +++ b/regression-test/suites/prepared_stmt_p0/cursor_fetch_empty_result.groovy @@ -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. + +suite("cursor_fetch_empty_result") { + String url = getServerPrepareJdbcUrl(context.config.jdbcUrl, "regression_test_prepared_stmt_p0") + + "&useCursorFetch=true&defaultFetchSize=10000&socketTimeout=10000" + + connect(context.config.jdbcUser, context.config.jdbcPassword, url) { + // With a positive defaultFetchSize Connector/J also converts a plain Statement into a + // server-prepared cursor execution, which is how BI tools commonly enter this path. + assertEquals(0, sql("SELECT 1 AS c WHERE 1 = 2").size()) + assertEquals([[1]], sql("SELECT 1 AS c WHERE 1 = 1")) + + def emptyResult = prepareStatement "SELECT 1 AS c WHERE 1 = 2" + assertEquals(com.mysql.cj.jdbc.ServerPreparedStatement, emptyResult.class) + qe_empty_result emptyResult + emptyResult.close() + + def nonEmptyResult = prepareStatement "SELECT 1 AS c WHERE 1 = 1" + assertEquals(com.mysql.cj.jdbc.ServerPreparedStatement, nonEmptyResult.class) + qe_non_empty_result nonEmptyResult + nonEmptyResult.close() + } + + String unidentifiedClientUrl = getServerPrepareJdbcUrl( + context.config.jdbcUrl, "regression_test_prepared_stmt_p0") + + "&useCursorFetch=true&defaultFetchSize=10000&connectionAttributes=none&socketTimeout=10000" + connect(context.config.jdbcUser, context.config.jdbcPassword, unidentifiedClientUrl) { + test { + sql "SELECT 1 AS c WHERE 1 = 2" + exception "Cannot safely execute cursor fetch because the client did not provide identifiable " + + "connection attributes" + } + } +}