Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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<String> 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<String, String> 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;
}
}
15 changes: 5 additions & 10 deletions fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlOkPacket.java
Original file line number Diff line number Diff line change
Expand Up @@ -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()) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Read the negotiated EOF capability here

This condition is false in the new unit test, but it remains true for real legacy-EOF connections. MysqlProto.negotiate records the client's bit only in MysqlChannel, then sets the serializer capability to context.getServerCapability(); that default mask always includes CLIENT_DEPRECATE_EOF. ProxyMysqlChannel starts with the same default as well. Consequently an authenticated client that did not negotiate the flag still gets the trailing zero byte this change intends to remove. Please key this from the negotiated/channel capability, and propagate it to proxy serialization, or store the negotiated mask in the serializer, with a handshake-level test.

// Connector/J parses the info field for CLIENT_DEPRECATE_EOF even when it is empty.
serializer.writeVInt(0);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ public class MysqlResultSetEndPacket extends MysqlPacket {

public MysqlResultSetEndPacket(QueryState state) {
this.serverStatus = state.serverStatus;
this.warningCount = state.getWarningRows();
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -509,6 +512,14 @@ public void setConnectAttributes(Map<String, String> 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();
}
Expand Down Expand Up @@ -997,6 +1008,7 @@ public void clear() {
statementContext = null;
loadBackendSelectionDecision = null;
loadBackendSelectionHint = null;
cursorFetchRequested = false;
}

// Arrow Flight SQL only.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Classify successful old-master reads without widening this guard

A real forwarded SELECT cannot satisfy this gate: result producers finish with QueryState.setEof(), while proxyExecute assigns status 0 only to OK and maps successful EOF to 1105. The follower therefore replays the unsafe cursor packets; the unit test mocks the impossible combination of query buffers plus status 0. Simply accepting EOF here would also reject safe old-master COM_QUERY and Connector/J 9.5+ results because this predicate never checks cursor intent or the compatibility class. Please recognize real result-set success, scope rejection to requests that need the cursor shim, test through real proxyExecute construction, and ensure a non-final multi-statement sends the local ERR only once.

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);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Preserve the old master's complete OK result

This rolling-upgrade branch rebuilds a successful DML response with only affectedRows, discarding the warning count and info string already encoded in the master's final OK packet. A normal forwarded INSERT calls OlapInsertExecutor.setReturnInfo(), which reports filteredRows as warnings and includes label/status/txnId in info; through an old master this branch changes those to zero warnings and no message. Please preserve or decode all protocol-visible OK fields (or safely reuse the ordinary OK packet) and cover a response with nonzero warnings and nonempty info.

}
packet = getResultPacket();
} else {
executor.sendProxyQueryResult();
packet = executor.getOutputPacket();
}
} else {
executor.sendResultSet(resultSet);
packet = getResultPacket();
Expand Down Expand Up @@ -729,6 +744,8 @@ public TMasterOpResult proxyExecute(TMasterOpRequest request) throws TException
if (request.isSetClientDeprecatedEOF() && request.isClientDeprecatedEOF()) {
ctx.getMysqlChannel().setClientDeprecatedEOF();
}
ctx.setCursorFetchRequested(request.isSetCursorFetchRequested()

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Handle cursor intent from old forwarding FEs

During a rolling upgrade an old follower cannot set the new optional cursor_fetch_requested field, so this silently records false. It still forwards CLIENT_DEPRECATE_EOF, Connector/J attributes, and, for parameterized statements, the execute buffer; the new master then emits binary rows but omits the compatibility metadata marker. An affected Connector/J cursor SELECT forwarded through that old FE can therefore still consume the final marker and hang. Please treat an absent cursor-intent field as an explicit mixed-version/unknown execute mode and fail safely when the affected combination cannot be disambiguated, with an old-sender/new-master parameterized cursor test.

&& request.isCursorFetchRequested());

ctx.setThreadLocalInfo();
StmtExecutor executor = null;
Expand Down Expand Up @@ -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);
Expand Down
15 changes: 15 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,7 @@ protected TMasterOpRequest buildStmtForwardParams() throws AnalysisException {
if (null != ctx.getPrepareExecuteBuffer()) {
params.setPrepareExecuteBuffer(ctx.getPrepareExecuteBuffer());
}
params.setCursorFetchRequested(ctx.isCursorFetchRequested());

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve binary execute mode without parameters

For a zero-placeholder prepared SELECT, handleExecute never sets prepareExecuteBuffer, so this forwards cursor_fetch_requested=true without any marker that makes the master take its prepared-execute branch. proxyExecute then leaves the fresh context as COM_SLEEP; both coordinator paths set mysql_row_binary_format=false, and non-empty rows are serialized as text even though Connector/J is reading a COM_STMT_EXECUTE binary result. The new regression uses getServerPrepareJdbcUrl, which connects directly to the master, so it misses this path. Please forward execute/binary-result intent independently of parameter bytes (or always carry an empty execute buffer) and cover a follower-to-master non-empty result.

}

ctx.getSessionContext().getDelegatedCredential().ifPresent((DelegatedCredential credential) -> {
Expand Down Expand Up @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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);
}
Expand Down
47 changes: 33 additions & 14 deletions fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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}.
*
Expand Down Expand Up @@ -1843,15 +1857,7 @@ private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo> 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<PrimitiveType> exprToStringType(List<Expr> exprs) {
Expand Down Expand Up @@ -1987,17 +1993,30 @@ private void sendFields(List<String> colNames, List<FieldInfo> 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);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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));
}
}
Loading
Loading