This is an automated email from the ASF dual-hosted git repository.
CalvinKirs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6fafa3195ee [fix](fe) Fix prepared statement streaming, forwarding and
cursor framing (#69314)
6fafa3195ee is described below
commit 6fafa3195ee4e90ddfb4d2b9c8e424e693d60512
Author: Calvin Kirs <[email protected]>
AuthorDate: Tue Oct 20 15:32:30 2026 +0800
[fix](fe) Fix prepared statement streaming, forwarding and cursor framing
(#69314)
### What problem does this PR solve?
Problem Summary:
- `COM_STMT_SEND_LONG_DATA` incorrectly returns an error packet, leaving
streaming clients out of sync. Accept parameter chunks without replying,
defer upload errors until EXECUTE, preserve binary bytes, and clear
upload state on EXECUTE/RESET/CLOSE.
- Reusing server-side prepared statements through a non-master FE can
fail with `NoSuchElementException: No value present`: the driver omits
unchanged parameter types, while the master prepares each forwarded
execution independently. Include cached types, the NULL bitmap and
complete values in forwarded EXECUTE payloads.
- Forwarded PREPARE can use a stale statement context when planning
placeholders. Use the parser's current prepare-stage context.
- Empty results with `useCursorFetch=true` can hang or leave unread
packets when client connection attributes are disabled or rewritten.
Return a valid empty cursor, complete FETCH with `LAST_ROW_SENT`, retain
cursor state on the client-facing FE, and preserve existing framing for
older FEs. Nonempty results continue to use buffered delivery; nonempty
cursor pagination is not added.
- Normalized streamed values can remain attached to idle pooled
connections after statement close/reset. Release the command-only
execution buffer after response finalization and forwarding complete.
### Release note
- Fix streamed parameters, repeated prepared-statement forwarding and
empty cursor results in the MySQL protocol, and reduce retained
parameter-buffer memory on idle connections.
### Check List (For Author)
- Test:
- [x] Unit Test: 32 focused long-data, protocol adapter and forwarding
tests passed on the final source, including the new buffer-lifecycle
test.
- [x] Regression test: `prepared_stmt_p0/send_long_data` and
`prepared_stmt_p0/cursor_fetch_empty_result` passed during
implementation.
- [x] Manual test: raw-protocol upload/cursor lifecycle and connection
reuse; rewritten client-attribute proxy reproduction passed after the
fix.
- [x] Checkstyle: zero violations in the latest modified Java files.
- Behavior changed:
- [x] Yes. Long-data uploads no longer produce responses; forwarded
parameters are self-contained; empty cursors finish through FETCH;
completed commands release execution buffers.
- Does this need documentation?
- [x] No.
---
.../doris/mysql/protocol/MysqlProtocolAdapter.java | 47 ++-
.../doris/mysql/protocol/MysqlResultSender.java | 15 +-
.../nereids/trees/expressions/literal/Literal.java | 6 +
.../java/org/apache/doris/qe/ConnectProcessor.java | 5 +-
.../org/apache/doris/qe/MysqlConnectProcessor.java | 108 +++++-
.../apache/doris/qe/PreparedStatementContext.java | 62 ++++
.../java/org/apache/doris/qe/StmtExecutor.java | 4 +-
.../mysql/protocol/MysqlProtocolAdapterTest.java | 42 +++
.../mysql/protocol/MysqlResultSenderTest.java | 37 ++-
.../doris/qe/FEOpExecutorMysqlProtocolTest.java | 2 +
.../qe/MysqlConnectProcessorCursorFetchTest.java | 24 ++
.../qe/MysqlConnectProcessorLongDataTest.java | 364 +++++++++++++++++++++
gensrc/thrift/FrontendService.thrift | 2 +
.../data/prepared_stmt_p0/send_long_data.out | 40 +++
.../suites/prepared_stmt_p0/send_long_data.groovy | 84 +++++
15 files changed, 808 insertions(+), 34 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapter.java
b/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapter.java
index a65bb00d95a..ad044cc7d19 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapter.java
@@ -35,6 +35,7 @@ import org.apache.doris.nereids.stats.StatsErrorEstimator;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.ConnectContext.ConnectType;
import org.apache.doris.qe.ConnectProcessor;
+import org.apache.doris.qe.PreparedStatementContext;
import org.apache.doris.qe.QueryState.MysqlStateType;
import org.apache.doris.qe.ShowResultSet;
import org.apache.doris.qe.StmtExecutor;
@@ -74,6 +75,9 @@ public class MysqlProtocolAdapter implements ProtocolAdapter {
private ByteBuffer prepareExecuteBuffer;
// Whether the current COM_STMT_EXECUTE requested a server-side read-only
cursor.
private boolean cursorFetchRequested;
+ private boolean resultHasRows;
+ private boolean emptyCursorSupported = true;
+ private PreparedStatementContext executingPreparedStatement;
public MysqlProtocolAdapter(MysqlChannel channel) {
this.channel = channel;
@@ -146,7 +150,7 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
*/
@Override
public void beforeStatement(ConnectContext ctx) {
- channel.reset();
+ resetResult();
}
@Override
@@ -183,6 +187,7 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
request.setPrepareExecuteBuffer(prepareExecuteBuffer);
}
request.setCursorFetchRequested(cursorFetchRequested);
+ request.setEmptyCursorSupported(true);
}
request.setClientDeprecatedEOF(channel.clientDeprecatedEOF());
request.setMysqlCapability(capability.getFlags());
@@ -212,6 +217,7 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
channel.setClientDeprecatedEOF();
}
cursorFetchRequested = request.isSetCursorFetchRequested() &&
request.isCursorFetchRequested();
+ emptyCursorSupported = request.isSetEmptyCursorSupported() &&
request.isEmptyCursorSupported();
}
/**
@@ -277,6 +283,13 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
return;
}
+ if (executingPreparedStatement != null) {
+ // A forwarded EXECUTE's final packet was encoded by the master.
Retain the cursor on
+ // the client-facing FE so its later FETCH can complete locally
without another RPC.
+ boolean cursorOpen =
Byte.toUnsignedInt(packet.get(packet.position())) == 0xFE
+ && (packet.get(packet.position() + 3) &
MysqlServerStatusFlag.SERVER_STATUS_CURSOR_EXISTS) != 0;
+ executingPreparedStatement.setEmptyCursorOpen(cursorOpen);
+ }
LOG.debug("Send to mysql channel for query {}",
DebugUtil.printId(ctx.queryId()));
channel.sendAndFlush(packet);
// note(wb) we should write profile after return result to mysql client
@@ -304,6 +317,11 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
// instead of the traditional EOF packet. This is required by the
MySQL protocol
// and expected by MySQL Connector/J 9.5.0+.
if (ctx.getState().getStateType() == MysqlStateType.EOF &&
channel.clientDeprecatedEOF()) {
+ if (emptyCursorSupported && cursorFetchRequested &&
!resultHasRows) {
+ // Both old and new drivers consume a cursor's opening
terminator. An empty
+ // cursor needs no retained rows and its first FETCH sends
LAST_ROW_SENT.
+ ctx.getState().serverStatus |=
MysqlServerStatusFlag.SERVER_STATUS_CURSOR_EXISTS;
+ }
packet = new MysqlResultSetEndPacket(ctx.getState());
} else {
packet = ctx.getState().toResponsePacket();
@@ -323,7 +341,9 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
@Override
public void afterStatement(ConnectContext ctx) {
+ prepareExecuteBuffer = null;
cursorFetchRequested = false;
+ executingPreparedStatement = null;
}
@Override
@@ -375,18 +395,31 @@ public class MysqlProtocolAdapter implements
ProtocolAdapter {
this.cursorFetchRequested = cursorFetchRequested;
}
- /**
- * Whether the client will consume the first OK packet after the column
definitions of the
- * result it asked a cursor for. Connector/J before 9.5 does, while
probing whether the cursor
- * was created; Doris never creates one, so the result must carry one more
end marker or an
- * empty result would leave the client waiting.
- */
+ /** Whether legacy buffered results need an extra terminator for this
client. */
public boolean clientConsumesCursorMetadataTerminator(ConnectContext ctx) {
return cursorFetchRequested
&&
MysqlCursorFetchCompatibility.resolve(ctx.getConnectAttributes())
!= MysqlCursorFetchCompatibility.Behavior.STANDARD;
}
+ public boolean isEmptyCursorSupported() {
+ return emptyCursorSupported;
+ }
+
+ public void beginPreparedExecute(PreparedStatementContext prepared) {
+ executingPreparedStatement = prepared;
+ prepared.setEmptyCursorOpen(false);
+ }
+
+ void recordResultRow() {
+ resultHasRows = true;
+ }
+
+ void resetResult() {
+ resultHasRows = false;
+ channel.reset();
+ }
+
// The xnio read loop. A command is read and processed while reads are
suspended, which is what
// serializes the commands of one connection.
public void startAcceptQuery(ConnectContext ctx, ConnectProcessor
connectProcessor) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlResultSender.java
b/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlResultSender.java
index d4327d117a9..66131c5e24a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlResultSender.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlResultSender.java
@@ -135,13 +135,14 @@ public class MysqlResultSender implements ResultSender {
@Override
public void sendRow(ByteBuffer row) throws IOException {
+ adapter.recordResultRow();
channel().sendOnePacket(row);
}
/** Clears the send flag and whatever a failed attempt of the query left
in the send buffer. */
@Override
public void reset() {
- channel().reset();
+ adapter.resetResult();
}
/**
@@ -270,10 +271,8 @@ public class MysqlResultSender implements ResultSender {
serializer.reset();
new MysqlEofPacket(ctx.getState()).writeTo(serializer);
channel.sendOnePacket(serializer.toByteBuffer());
- } else if (adapter.clientConsumesCursorMetadataTerminator(ctx)) {
- // 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.
+ } else if (!adapter.isEmptyCursorSupported() &&
adapter.clientConsumesCursorMetadataTerminator(ctx)) {
+ // Preserve framing for older client-facing FEs during rolling
upgrades.
serializer.reset();
new MysqlResultSetEndPacket(ctx.getState()).writeTo(serializer);
channel.sendOnePacket(serializer.toByteBuffer());
@@ -282,7 +281,6 @@ public class MysqlResultSender implements ResultSender {
private void sendTextResultRow(ResultSet resultSet) throws IOException {
MysqlSerializer serializer = serializer();
- MysqlChannel channel = channel();
for (List<String> row : resultSet.getResultRows()) {
serializer.reset();
for (String item : row) {
@@ -292,13 +290,12 @@ public class MysqlResultSender implements ResultSender {
serializer.writeLenEncodedString(item);
}
}
- channel.sendOnePacket(serializer.toByteBuffer());
+ sendRow(serializer.toByteBuffer());
}
}
private void sendBinaryResultRow(ResultSet resultSet) throws IOException {
MysqlSerializer serializer = serializer();
- MysqlChannel channel = channel();
//
https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_binary_resultset.html#sect_protocol_binary_resultset_row_value
ResultSetMetaData metaData = resultSet.getMetaData();
int nullBitmapLength = (metaData.getColumnCount() + 7 + 2) / 8;
@@ -361,7 +358,7 @@ public class MysqlResultSender implements ResultSender {
}
}
}
- channel.sendOnePacket(serializer.toByteBuffer());
+ sendRow(serializer.toByteBuffer());
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/literal/Literal.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/literal/Literal.java
index 92de0c55f1f..dfeb493fcaf 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/literal/Literal.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/literal/Literal.java
@@ -591,6 +591,12 @@ public abstract class Literal extends Expression
implements LeafExpression {
case MYSQL_TYPE_VARCHAR:
literal = handleVarcharLiteral(data);
break;
+ case MYSQL_TYPE_TINY_BLOB:
+ case MYSQL_TYPE_MEDIUM_BLOB:
+ case MYSQL_TYPE_LONG_BLOB:
+ case MYSQL_TYPE_BLOB:
+ literal = new
VarBinaryLiteral(MysqlProto.readLenEncodedString(data));
+ break;
default:
throw new AnalysisException("Unsupported MySQL type: " +
mysqlType);
}
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 c1ad7ffd03f..f299bb6a677 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
@@ -156,11 +156,14 @@ public abstract class ConnectProcessor {
}
protected void handleStmtResetById(int stmtId) {
- if (ctx.getPreparedStementContext(String.valueOf(stmtId)) == null) {
+ PreparedStatementContext prepared =
ctx.getPreparedStementContext(String.valueOf(stmtId));
+ if (prepared == null) {
ctx.getState().setError(ErrorCode.ERR_UNKNOWN_STMT_HANDLER,
String.format("Unknown prepared statement handler (%s)
given to mysqld_stmt_reset", stmtId));
return;
}
+ prepared.clearLongData();
+ prepared.setEmptyCursorOpen(false);
ctx.getState().setOk();
}
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 af8c804259c..1486d9c210e 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
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MysqlColType;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.AuthenticationException;
import org.apache.doris.common.ConnectionException;
import org.apache.doris.common.ErrorCode;
@@ -35,6 +36,7 @@ import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.mysql.MysqlHandshakePacket;
import org.apache.doris.mysql.MysqlProto;
import org.apache.doris.mysql.MysqlSerializer;
+import org.apache.doris.mysql.MysqlServerStatusFlag;
import org.apache.doris.mysql.authenticate.password.NativePasswordResolver;
import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.mysql.protocol.MysqlProtocolAdapter;
@@ -93,6 +95,27 @@ public class MysqlConnectProcessor extends ConnectProcessor {
handleStmtResetById(stmtId);
}
+ private void handleSendLongData() {
+ // This command never has a response, including when the upload fails.
Report a known
+ // statement's upload error on EXECUTE instead, otherwise the
connection gets out of sync.
+ ctx.getState().setNoop();
+ packetBuf.order(ByteOrder.LITTLE_ENDIAN);
+ if (packetBuf.remaining() < Integer.BYTES) {
+ return;
+ }
+ int stmtId = packetBuf.getInt();
+ PreparedStatementContext prepared =
ctx.getPreparedStementContext(String.valueOf(stmtId));
+ if (prepared == null) {
+ return;
+ }
+ if (packetBuf.remaining() < Short.BYTES) {
+ prepared.setLongDataError("Missing parameter index in
COM_STMT_SEND_LONG_DATA");
+ return;
+ }
+ int parameter = Short.toUnsignedInt(packetBuf.getShort());
+ prepared.appendLongData(parameter, packetBuf,
ctx.getSessionVariable().getMaxAllowedPacket());
+ }
+
private String getPacket() {
byte[] bytes = packetBuf.array();
StringBuilder printB = new StringBuilder();
@@ -126,11 +149,16 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
@Override
protected void handleExecute(PrepareCommand prepareCommand, long stmtId,
PreparedStatementContext prepCtx,
ByteBuffer packetBuf, TUniqueId queryId) {
+ MysqlProtocolAdapter.of(ctx).beginPreparedExecute(prepCtx);
int paramCount = prepareCommand.placeholderCount();
LOG.debug("execute prepared statement {}, paramCount {}", stmtId,
paramCount);
// null bitmap
String stmtStr = "";
try {
+ if (prepCtx.getLongDataError() != null) {
+ ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR,
prepCtx.getLongDataError());
+ return;
+ }
ctx.getProtocolAdapter().beforeStatement(ctx);
StatementContext statementContext = prepCtx.getStatementContext();
if (!ctx.isProxy()) {
@@ -144,7 +172,8 @@ public class MysqlConnectProcessor extends ConnectProcessor
{
byte[] nullbitmapData = new byte[(paramCount + 7) / 8];
packetBuf.get(nullbitmapData);
// new_params_bind_flag
- if ((int) packetBuf.get() != 0) {
+ boolean hasNewParameterTypes = packetBuf.get() != 0;
+ if (hasNewParameterTypes) {
List<Placeholder> typedPlaceholders = new ArrayList<>();
// parse params's types
for (int i = 0; i < paramCount; ++i) {
@@ -158,6 +187,22 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
prepCtx.command =
prepareCommand.withPlaceholders(typedPlaceholders);
prepareCommand = (PrepareCommand) prepCtx.command;
}
+ // The master prepares each forwarded EXECUTE again without
connection-local state.
+ // Include cached parameter types and inline uploaded values
so it can bind independently.
+ MysqlSerializer forwarded = prepCtx.hasLongData() ||
!hasNewParameterTypes
+ ? MysqlSerializer.newInstance() : null;
+ if (forwarded != null) {
+ for (int i = 0; i < paramCount; ++i) {
+ if (prepCtx.hasLongData(i)) {
+ nullbitmapData[i / 8] &= ~(1 << (i & 7));
+ }
+ }
+ forwarded.writeBytes(nullbitmapData);
+ forwarded.writeInt1(1);
+ for (Placeholder placeholder :
prepareCommand.getPlaceholders()) {
+ forwarded.writeInt2(placeholder.getMysqlTypeCode());
+ }
+ }
// parse param data
for (int i = 0; i < paramCount; ++i) {
PlaceholderId exprId =
prepareCommand.getPlaceholders().get(i).getPlaceholderId();
@@ -168,9 +213,38 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
}
MysqlColType type =
prepareCommand.getPlaceholders().get(i).getMysqlColType();
boolean isUnsigned =
prepareCommand.getPlaceholders().get(i).isUnsigned();
- Literal l = Literal.getLiteralByMysqlType(type,
isUnsigned, packetBuf);
+ byte[] longData = prepCtx.getLongData(i);
+ Literal l;
+ if (longData != null) {
+ switch (type) {
+ case MYSQL_TYPE_STRING:
+ case MYSQL_TYPE_VARSTRING:
+ case MYSQL_TYPE_VARCHAR:
+ case MYSQL_TYPE_TINY_BLOB:
+ case MYSQL_TYPE_MEDIUM_BLOB:
+ case MYSQL_TYPE_LONG_BLOB:
+ case MYSQL_TYPE_BLOB:
+ break;
+ default:
+ throw new AnalysisException("Invalid MySQL
type for long data: " + type);
+ }
+ MysqlSerializer value = MysqlSerializer.newInstance();
+ value.writeLenEncodedBytes(longData);
+ l = Literal.getLiteralByMysqlType(type, isUnsigned,
value.toByteBuffer());
+ forwarded.writeBytes(value.toArray());
+ } else {
+ int start = packetBuf.position();
+ l = Literal.getLiteralByMysqlType(type, isUnsigned,
packetBuf);
+ if (forwarded != null) {
+ forwarded.writeBytes(packetBuf.array(),
packetBuf.arrayOffset() + start,
+ packetBuf.position() - start);
+ }
+ }
statementContext.getIdToPlaceholderRealExpr().put(exprId,
l);
}
+ if (forwarded != null && !ctx.isProxy()) {
+ ctx.setPrepareExecuteBuffer(forwarded.toByteBuffer());
+ }
}
ExecuteCommand executeStmt = new
ExecuteCommand(String.valueOf(stmtId), prepareCommand, statementContext);
// TODO set real origin statement
@@ -196,8 +270,10 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
LOG.warn("Process one query failed because unknown reason: ", e);
ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR,
e.getClass().getSimpleName() + ", msg: " + e.getMessage());
+ } finally {
+ prepCtx.clearLongData();
}
- if (ctx.getSessionVariable().isEnablePreparedStmtAuditLog()) {
+ if (ctx.getSessionVariable().isEnablePreparedStmtAuditLog() &&
executor != null) {
auditAfterExec(stmtStr, executor.getParsedStmt(),
executor.getQueryStatisticsForAuditLog(), true);
} else {
// When audit log is disabled for prepared statements, still
update QPS metrics.
@@ -234,6 +310,23 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
handleExecute(preparedStatementContext.command, stmtId,
preparedStatementContext, packetBuf, null);
}
+ protected void handleStmtFetch(int stmtId) {
+ PreparedStatementContext prepared =
ctx.getPreparedStementContext(String.valueOf(stmtId));
+ if (prepared == null) {
+ ctx.getState().setError(ErrorCode.ERR_UNKNOWN_STMT_HANDLER,
+ String.format("Unknown prepared statement handler (%s)
given to mysqld_stmt_fetch", stmtId));
+ return;
+ }
+ if (!prepared.isEmptyCursorOpen()) {
+ ctx.getState().setError(ErrorCode.ERR_STMT_HAS_NO_OPEN_CURSOR,
+ String.format("The statement (%d) has no open cursor.",
stmtId));
+ return;
+ }
+ prepared.setEmptyCursorOpen(false);
+ ctx.getState().setEof();
+ ctx.getState().serverStatus |=
MysqlServerStatusFlag.SERVER_STATUS_LAST_ROW_SENT;
+ }
+
// Process COM_QUERY statement,
private void handleQuery() throws ConnectionException {
// convert statement to Java string
@@ -278,6 +371,15 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
case COM_STMT_EXECUTE:
handleExecute();
break;
+ case COM_STMT_FETCH:
+ packetBuf.order(ByteOrder.LITTLE_ENDIAN);
+ int fetchStmtId = packetBuf.getInt();
+ packetBuf.getInt(); // Requested row count; an empty cursor
has no rows to return.
+ handleStmtFetch(fetchStmtId);
+ break;
+ case COM_STMT_SEND_LONG_DATA:
+ handleSendLongData();
+ break;
case COM_FIELD_LIST:
handleFieldList();
break;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/PreparedStatementContext.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/PreparedStatementContext.java
index f28f6bd1c53..e16dd5fb7ab 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/PreparedStatementContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/PreparedStatementContext.java
@@ -21,6 +21,10 @@ import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
import org.apache.doris.planner.GroupCommitPlanner;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Optional;
public class PreparedStatementContext {
@@ -30,6 +34,10 @@ public class PreparedStatementContext {
public String stmtString;
public Optional<ShortCircuitQueryContext> shortCircuitQueryContext =
Optional.empty();
public Optional<GroupCommitPlanner> groupCommitPlanner = Optional.empty();
+ // Long data belongs to a prepared statement and survives individual
upload commands.
+ private final Map<Integer, ByteArrayOutputStream> longData = new
HashMap<>();
+ private String longDataError;
+ private boolean emptyCursorOpen;
// Timestamp in millisecond last command starts at
protected volatile long startTime;
@@ -42,6 +50,14 @@ public class PreparedStatementContext {
this.stmtString = stmtString;
}
+ public boolean isEmptyCursorOpen() {
+ return emptyCursorOpen;
+ }
+
+ public void setEmptyCursorOpen(boolean emptyCursorOpen) {
+ this.emptyCursorOpen = emptyCursorOpen;
+ }
+
public long getStartTime() {
return startTime;
}
@@ -81,4 +97,50 @@ public class PreparedStatementContext {
public void setStartTime() {
startTime = System.currentTimeMillis();
}
+
+ public void appendLongData(int parameter, ByteBuffer data, int
maxAllowedPacket) {
+ if (longDataError != null) {
+ return;
+ }
+ if (parameter >= command.placeholderCount()) {
+ setLongDataError("Invalid parameter index " + parameter + " in
COM_STMT_SEND_LONG_DATA");
+ return;
+ }
+ ByteArrayOutputStream value = longData.computeIfAbsent(parameter, key
-> new ByteArrayOutputStream());
+ if ((long) value.size() + data.remaining() > maxAllowedPacket) {
+ setLongDataError("Parameter data in COM_STMT_SEND_LONG_DATA
exceeds max_allowed_packet ("
+ + maxAllowedPacket + ")");
+ return;
+ }
+ byte[] chunk = new byte[data.remaining()];
+ data.get(chunk);
+ value.write(chunk, 0, chunk.length);
+ }
+
+ public boolean hasLongData() {
+ return !longData.isEmpty();
+ }
+
+ public boolean hasLongData(int parameter) {
+ return longData.containsKey(parameter);
+ }
+
+ public byte[] getLongData(int parameter) {
+ ByteArrayOutputStream value = longData.get(parameter);
+ return value == null ? null : value.toByteArray();
+ }
+
+ public String getLongDataError() {
+ return longDataError;
+ }
+
+ public void setLongDataError(String error) {
+ longData.clear();
+ longDataError = error;
+ }
+
+ public void clearLongData() {
+ longData.clear();
+ longDataError = null;
+ }
}
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 c2a0f689b4b..2506e065906 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
@@ -2170,7 +2170,9 @@ public class StmtExecutor {
Preconditions.checkState(parsedStmt instanceof LogicalPlanAdapter,
"Nereids only process LogicalPlanAdapter,"
+ " but parsedStmt is " +
parsedStmt.getClass().getName());
- NereidsPlanner nereidsPlanner = new
PrepareCommandPlanner(statementContext);
+ // Proxy parsing creates a new context. Use the context marked as the
prepare stage
+ // by PrepareCommand, rather than the one allocated before parsing in
the constructor.
+ NereidsPlanner nereidsPlanner = new
PrepareCommandPlanner(context.getStatementContext());
nereidsPlanner.plan(parsedStmt,
context.getSessionVariable().toThrift());
return
nereidsPlanner.getCascadesContext().getRewritePlan().getOutput();
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapterTest.java
index 22886f63941..69fdc66213f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapterTest.java
@@ -28,10 +28,13 @@ import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.mysql.MysqlProto;
import org.apache.doris.mysql.MysqlServerStatusFlag;
import org.apache.doris.mysql.ProxyMysqlChannel;
+import org.apache.doris.nereids.StatementContext;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.ConnectContext.ConnectType;
import org.apache.doris.qe.ConnectProcessor;
import org.apache.doris.qe.ConnectScheduler;
+import org.apache.doris.qe.PreparedStatementContext;
+import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.qe.protocol.RecordingMysqlChannel;
import org.apache.doris.thrift.TMasterOpRequest;
import org.apache.doris.thrift.TResultSinkType;
@@ -137,6 +140,8 @@ public class MysqlProtocolAdapterTest {
MysqlProtocolAdapter protocol = MysqlProtocolAdapter.of(ctx);
ctx.setConnectAttributes(ImmutableMap.of("_client_name", "MySQL
Connector/J", "_client_version", "8.2.0"));
+ // An older FE cannot handle FETCH. Preserve its legacy framing during
an upgrade.
+ protocol.restoreFromForwardRequest(ctx, new TMasterOpRequest());
// Only a COM_STMT_EXECUTE that asked for a cursor ...
Assertions.assertFalse(protocol.clientConsumesCursorMetadataTerminator(ctx));
ctx.setCursorFetchRequested(true);
@@ -151,6 +156,42 @@ public class MysqlProtocolAdapterTest {
Assertions.assertFalse(ctx.isCursorFetchRequested());
}
+ @Test
+ public void testEmptyCursorOpeningAndForwardedCursorState() throws
Exception {
+ RecordingMysqlChannel channel = new RecordingMysqlChannel();
+ channel.setClientDeprecatedEOF();
+ ConnectContext ctx = new ConnectContext(new
MysqlProtocolAdapter(channel));
+ MysqlProtocolAdapter protocol = MysqlProtocolAdapter.of(ctx);
+ ctx.setCursorFetchRequested(true);
+ ctx.getState().setEof();
+ PreparedStatementContext prepared = new PreparedStatementContext(
+ null, ctx, new StatementContext(), "7");
+ protocol.beginPreparedExecute(prepared);
+ protocol.finishCommand(ctx, null);
+ Assertions.assertTrue(prepared.isEmptyCursorOpen());
+
Assertions.assertEquals(MysqlServerStatusFlag.SERVER_STATUS_CURSOR_EXISTS,
+ ctx.getState().serverStatus &
MysqlServerStatusFlag.SERVER_STATUS_CURSOR_EXISTS);
+
+ // A row, including a binary NULL row, means this is an ordinary
result, not an empty cursor.
+ protocol.beforeStatement(ctx);
+ ctx.getState().reset();
+ ctx.getState().setEof();
+ protocol.beginPreparedExecute(prepared);
+ ctx.getResultSender().sendRow(ByteBuffer.wrap(new byte[] {0, 4}));
+ protocol.finishCommand(ctx, null);
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+
+ // The master sends CURSOR_EXISTS in its final packet; retain it on
the client-facing FE.
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ Mockito.when(executor.hasForwardedToMaster()).thenReturn(true);
+ Mockito.when(executor.getOutputPacket()).thenReturn(ByteBuffer.wrap(
+ new byte[] {(byte) 0xFE, 0, 0, 0x40, 0, 0, 0, 0}));
+ protocol.finishCommand(ctx, executor);
+ Assertions.assertTrue(prepared.isEmptyCursorOpen());
+ protocol.beginPreparedExecute(prepared);
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+ }
+
@Test
public void testClientAddressComesFromTheChannel() {
MysqlChannel channel = Mockito.mock(MysqlChannel.class);
@@ -329,6 +370,7 @@ public class MysqlProtocolAdapterTest {
Assertions.assertEquals(flags, request.getMysqlCapability());
Assertions.assertArrayEquals(new byte[] {7, 0, 0, 0},
request.getPrepareExecuteBuffer());
Assertions.assertTrue(request.isCursorFetchRequested());
+ Assertions.assertTrue(request.isEmptyCursorSupported());
// The master's proxy context takes them over.
ConnectContext proxy = ConnectContext.forMysqlProxy("session-1");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlResultSenderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlResultSenderTest.java
index 2a91ae1feb8..c9a333c5c7f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlResultSenderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mysql/protocol/MysqlResultSenderTest.java
@@ -33,6 +33,7 @@ import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.ResultSet;
import org.apache.doris.qe.protocol.RecordingMysqlChannel;
import org.apache.doris.qe.protocol.RecordingMysqlChannel.RecordedPacket;
+import org.apache.doris.thrift.TMasterOpRequest;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
@@ -154,25 +155,35 @@ public class MysqlResultSenderTest {
}
@Test
- public void testCursorFetchMetadataTerminatorDependsOnConnectorJVersion()
throws IOException {
- List<byte[]> 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());
+ public void testCursorFetchMetadataHasNoExtraTerminator() throws
IOException {
+ for (String version : new String[] {"8.0.27", "8.4.0", "9.4.0",
"9.7.0"}) {
+ Assertions.assertEquals(2, sendEmptyResultSet(true, "MySQL
Connector/J", version).size());
+ }
+ Assertions.assertEquals(2, sendEmptyResultSet(true,
Collections.emptyMap(), true).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(3, sendEmptyResultSet(true,
Collections.emptyMap(), true).size());
-
List<byte[]> 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 testOlderFeKeepsLegacyCursorMetadataFraming() throws
IOException {
+ RecordingMysqlChannel channel = newChannel(true);
+ ConnectContext ctx = newContext(channel);
+ TMasterOpRequest request = new TMasterOpRequest();
+ request.setClientDeprecatedEOF(true);
+ request.setCursorFetchRequested(true);
+ MysqlProtocolAdapter.of(ctx).restoreFromForwardRequest(ctx, request);
+ ctx.setConnectAttributes(ImmutableMap.of(
+ "_client_name", "MySQL Connector/J", "_client_version",
"8.0.27"));
+ ctx.getResultSender().sendResultSet(new CommonResultSet(
+ new CommonResultSetMetaData(Collections.singletonList(new
Column("c", PrimitiveType.INT))),
+ Collections.emptyList()), null, true);
+ List<byte[]> packets = payloads(channel);
+ Assertions.assertEquals(3, packets.size());
+ Assertions.assertEquals(0xFE, Byte.toUnsignedInt(packets.get(2)[0]));
+ }
+
@Test
public void testPrepareMetadataTerminatorsFollowNegotiatedCapability()
throws IOException {
Assertions.assertEquals(3, sendPrepareMetadata(false).size());
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
index f3a019badb3..fc2d0a26424 100644
---
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
@@ -61,6 +61,7 @@ public class FEOpExecutorMysqlProtocolTest {
Assertions.assertTrue(request.isClientDeprecatedEOF());
Assertions.assertTrue(request.isCursorFetchRequested());
+ Assertions.assertTrue(request.isEmptyCursorSupported());
Assertions.assertEquals("8.2.0",
request.getConnectAttributes().get("_client_version"));
for (byte[] payload : Arrays.asList(new byte[0], new byte[] {0, 1,
3, 0, 42, 0, 0, 0})) {
@@ -72,6 +73,7 @@ public class FEOpExecutorMysqlProtocolTest {
Assertions.assertTrue(restored.isSetPrepareExecuteBuffer());
Assertions.assertArrayEquals(payload,
restored.getPrepareExecuteBuffer());
Assertions.assertTrue(restored.isCursorFetchRequested());
+ Assertions.assertTrue(restored.isEmptyCursorSupported());
}
}
}
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
index 8f788d75ad4..1694ccb3185 100644
---
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
@@ -82,6 +82,30 @@ public class MysqlConnectProcessorCursorFetchTest {
"Unknown prepared statement handler"));
}
+ @Test
+ public void testEmptyCursorFetchResetAndClose() throws Exception {
+ ConnectContext context = new ConnectContext();
+ PreparedStatementContext prepared = new PreparedStatementContext(
+ Mockito.mock(PrepareCommand.class), context, new
StatementContext(), "7");
+ context.addPreparedStatementContext("7", prepared);
+ MysqlConnectProcessor processor = new MysqlConnectProcessor(context);
+ prepared.setEmptyCursorOpen(true);
+ processor.handleStmtFetch(7);
+ Assertions.assertEquals(QueryState.MysqlStateType.EOF,
context.getState().getStateType());
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+ Assertions.assertEquals(0x80, context.getState().serverStatus & 0xC0);
+
+ context.getState().reset();
+ processor.handleStmtFetch(7);
+ Assertions.assertEquals(1421,
context.getState().getErrorCode().getCode());
+ prepared.setEmptyCursorOpen(true);
+ processor.handleStmtResetById(7);
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+ context.removePrepareStmt("7");
+ processor.handleStmtFetch(7);
+ Assertions.assertEquals(1243,
context.getState().getErrorCode().getCode());
+ }
+
private ConnectContext execute(boolean cursorRequested, boolean
clientDeprecatedEof,
boolean identifiedClient) throws Exception {
ConnectContext context = new ConnectContext();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
new file mode 100644
index 00000000000..d649f469a7d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
@@ -0,0 +1,364 @@
+// 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.catalog.MysqlColType;
+import org.apache.doris.mysql.MysqlChannel;
+import org.apache.doris.mysql.MysqlSerializer;
+import org.apache.doris.mysql.protocol.MysqlProtocolAdapter;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.trees.expressions.Placeholder;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarBinaryLiteral;
+import org.apache.doris.nereids.trees.plans.PlaceholderId;
+import org.apache.doris.nereids.trees.plans.PrepareCommandPlanner;
+import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.stream.IntStream;
+
+public class MysqlConnectProcessorLongDataTest {
+ @Test
+ public void testUploadHasNoResponseAndPingStillWorks() throws Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_BLOB);
+ h.upload(7, 0, new byte[] {0, (byte) 0xff});
+ h.upload(7, 0, new byte[] {1});
+ h.upload(123, 0, new byte[] {2}); // Unknown statements must not
produce a response either.
+ Assertions.assertArrayEquals(new byte[] {0, (byte) 0xff, 1},
h.prepared.getLongData(0));
+ Mockito.verify(h.channel, Mockito.never()).sendAndFlush(Mockito.any());
+ h.command(ByteBuffer.wrap(new byte[] {(byte) 0x0e}));
+ Mockito.verify(h.channel).sendAndFlush(Mockito.any());
+ }
+
+ @Test
+ public void testExecuteReleasesNormalizedUploadAfterResponse() throws
Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_BLOB);
+ h.upload(7, 0, new byte[] {0, (byte) 0xff, (byte) 0x80, 65});
+ try (MockedConstruction<StmtExecutor> executors =
Mockito.mockConstruction(StmtExecutor.class,
+ (executor, construction) -> Mockito.doAnswer(invocation -> {
+ // The complete upload must remain available while
execution can forward it.
+ Assertions.assertArrayEquals(new byte[] {0, 1, (byte) 252,
0, 4, 0, (byte) 0xff, (byte) 0x80, 65},
+ h.context.getPrepareExecuteBuffer().array());
+ return null;
+ }).when(executor).execute());
+ MockedStatic<AuditLogHelper> audit =
Mockito.mockStatic(AuditLogHelper.class)) {
+ h.command(ByteBuffer.wrap(new byte[] {0x17, 7, 0, 0, 0, 0, 1, 0,
0, 0, 0, 1, (byte) 252, 0}));
+ Assertions.assertEquals(QueryState.MysqlStateType.OK,
h.context.getState().getStateType(),
+ h.context.getState().getErrorMessage());
+ Mockito.verify(executors.constructed().get(0)).execute();
+ Mockito.verify(h.channel).sendAndFlush(Mockito.any());
+ Assertions.assertFalse(h.prepared.hasLongData());
+ Assertions.assertNull(h.context.getPrepareExecuteBuffer());
+ }
+ }
+
+ @Test
+ public void testMixedParametersAndForwardingWithCachedTypes() throws
Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_BLOB,
MysqlColType.MYSQL_TYPE_LONG,
+ MysqlColType.MYSQL_TYPE_VARSTRING,
MysqlColType.MYSQL_TYPE_VARSTRING);
+ h.upload(7, 0, new byte[] {0, (byte) 0xff});
+ h.upload(7, 0, new byte[] {(byte) 0x80});
+ // Split one UTF-8 character across upload packets.
+ byte[] text = "中文".getBytes(StandardCharsets.UTF_8);
+ h.upload(7, 2, new byte[] {text[0]});
+ h.upload(7, 2, java.util.Arrays.copyOfRange(text, 1, text.length));
+ // The fourth parameter is NULL. Uploaded parameters have no inline
values.
+ h.execute(new byte[] {8, 0, 42, 0, 0, 0});
+ Assertions.assertEquals(new VarBinaryLiteral(new byte[] {0, (byte)
0xff, (byte) 0x80}), h.bound(0));
+ Assertions.assertEquals(new IntegerLiteral(42), h.bound(1));
+ Assertions.assertEquals(new StringLiteral("中文"), h.bound(2));
+ Assertions.assertInstanceOf(NullLiteral.class, h.bound(3));
+ Assertions.assertFalse(h.prepared.hasLongData());
+
+ assertForwardedParameters(h);
+ }
+
+ @Test
+ public void testForwardInlineParametersWithCachedTypes() throws Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_VARSTRING,
MysqlColType.MYSQL_TYPE_LONG,
+ MysqlColType.MYSQL_TYPE_VARSTRING);
+ MysqlSerializer initial = MysqlSerializer.newInstance();
+ initial.writeInt1(4); // Third parameter is NULL, but still has a
cached type.
+ initial.writeInt1(1);
+ initial.writeInt2(MysqlColType.MYSQL_TYPE_VARSTRING.getCode());
+ initial.writeInt2(MysqlColType.MYSQL_TYPE_LONG.getCode() | 0x8000);
+ initial.writeInt2(MysqlColType.MYSQL_TYPE_VARSTRING.getCode());
+ initial.writeLenEncodedString("first");
+ initial.writeInt4(-1);
+ h.execute(initial.toArray());
+ assertForwardedParameters(h);
+
+ MysqlSerializer cached = MysqlSerializer.newInstance();
+ cached.writeInt1(1); // First parameter becomes NULL; third becomes
non-NULL.
+ cached.writeInt1(0);
+ cached.writeInt4(-2);
+ cached.writeLenEncodedString("second");
+ h.execute(cached.toArray());
+ Assertions.assertInstanceOf(NullLiteral.class, h.bound(0));
+ Assertions.assertEquals(new StringLiteral("second"), h.bound(2));
+ assertForwardedParameters(h);
+ }
+
+ @Test
+ public void testForwardUpdatedTypesAndAllNullParameters() throws Exception
{
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_LONG);
+ h.execute(new byte[] {0, 1, 3, 0, 42, 0, 0, 0});
+ assertForwardedParameters(h);
+ h.execute(new byte[] {0, 0, 43, 0, 0, 0});
+ assertForwardedParameters(h);
+ h.execute(new byte[] {0, 1, (byte) 253, 0, 1, 65});
+ assertForwardedParameters(h);
+ h.execute(new byte[] {1, 0});
+ Assertions.assertInstanceOf(NullLiteral.class, h.bound(0));
+ assertForwardedParameters(h);
+ h.execute(new byte[] {0, 0, 1, 66});
+ Assertions.assertEquals(new StringLiteral("B"), h.bound(0));
+ assertForwardedParameters(h);
+ }
+
+ private static void assertForwardedParameters(Harness follower) throws
Exception {
+ ByteBuffer forwarded =
follower.context.getPrepareExecuteBuffer().duplicate();
+ byte[] payload = new byte[forwarded.remaining()];
+ forwarded.get(payload);
+ // A proxy EXECUTE starts with a fresh PREPARE, without the client's
cached parameter types.
+ Harness master = new
Harness(follower.prepared.command.placeholderCount());
+ for (Placeholder placeholder :
master.prepared.command.getPlaceholders()) {
+ Assertions.assertFalse(placeholder.hasMysqlColType());
+ }
+ master.execute(payload);
+ for (int i = 0; i < follower.prepared.command.placeholderCount(); ++i)
{
+ Assertions.assertEquals(follower.bound(i), master.bound(i));
+
Assertions.assertEquals(follower.prepared.command.getPlaceholders().get(i).getMysqlTypeCode(),
+
master.prepared.command.getPlaceholders().get(i).getMysqlTypeCode());
+ }
+ }
+
+ @Test
+ public void testEmptyUploadAndStatementIsolation() throws Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_VARSTRING);
+ PreparedStatementContext other = new
PreparedStatementContext(h.prepared.command, h.context,
+ new StatementContext(), "select ?");
+ h.context.addPreparedStatementContext("8", other);
+ h.upload(8, 0, new byte[] {65});
+ h.upload(7, 0, new byte[0]);
+ Assertions.assertTrue(h.prepared.hasLongData(0));
+ h.execute(new byte[] {0, 0});
+ Assertions.assertEquals(new StringLiteral(""), h.bound(0));
+ Assertions.assertArrayEquals(new byte[] {65}, other.getLongData(0));
+ // The next execution must consume its inline value instead of reusing
uploaded data.
+ h.execute(new byte[] {0, 0, 1, 66});
+ Assertions.assertEquals(new StringLiteral("B"), h.bound(0));
+ }
+
+ @Test
+ public void testUploadErrorsAreDeferredAndResetClearsThem() throws
Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_VARSTRING);
+ h.context.getSessionVariable().maxAllowedPacket = 3;
+ h.upload(7, 0, new byte[] {65, 66});
+ h.upload(7, 0, new byte[] {67, 68});
+ Assertions.assertNotNull(h.prepared.getLongDataError());
+ Assertions.assertFalse(h.prepared.hasLongData());
+ Mockito.verify(h.channel, Mockito.never()).sendAndFlush(Mockito.any());
+ h.executeError(new byte[] {0, 0}, "max_allowed_packet");
+ Assertions.assertNull(h.prepared.getLongDataError());
+ h.upload(7, 65535, new byte[0]);
+ Assertions.assertNotNull(h.prepared.getLongDataError());
+ h.processor.handleStmtResetById(7);
+ Assertions.assertNull(h.prepared.getLongDataError());
+ h.upload(7, 0, new byte[] {65});
+ h.processor.handleStmtResetById(7);
+ Assertions.assertFalse(h.prepared.hasLongData());
+ h.execute(new byte[] {0, 0, 1, 66});
+ Assertions.assertEquals(new StringLiteral("B"), h.bound(0));
+ }
+
+ @Test
+ public void testBindingFailureClearsUploadWithAuditEnabled() throws
Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_LONG);
+ h.context.getSessionVariable().enablePreparedStmtAuditLog = true;
+ h.upload(7, 0, new byte[] {65});
+ h.executeError(new byte[] {0, 0}, "Invalid MySQL type for long data");
+ Assertions.assertFalse(h.prepared.hasLongData());
+ }
+
+ @Test
+ public void testMalformedUploadAndCloseHaveNoResponse() throws Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_VARSTRING);
+ h.command(ByteBuffer.wrap(new byte[] {0x18}));
+ h.command(ByteBuffer.wrap(new byte[] {0x18, 7, 0, 0, 0}));
+ Assertions.assertNotNull(h.prepared.getLongDataError());
+ Mockito.verify(h.channel, Mockito.never()).sendAndFlush(Mockito.any());
+ h.processor.handleStmtResetById(7);
+ h.upload(7, 0, new byte[] {65});
+ h.context.setThreadLocalInfo();
+ try {
+ h.command(ByteBuffer.wrap(new byte[] {0x19, 7, 0, 0, 0}));
+ Assertions.assertNull(h.context.getPreparedStementContext("7"));
+ Mockito.verify(h.channel,
Mockito.never()).sendAndFlush(Mockito.any());
+ } finally {
+ ConnectContext.remove();
+ }
+ }
+
+ @Test
+ public void testUploadOverridesNullBitmap() throws Exception {
+ Harness h = new Harness(MysqlColType.MYSQL_TYPE_VARSTRING);
+ h.upload(7, 0, new byte[] {65});
+ h.execute(new byte[] {1, 0});
+ Assertions.assertEquals(new StringLiteral("A"), h.bound(0));
+ Assertions.assertEquals(0, h.context.getPrepareExecuteBuffer().get(0));
+ }
+
+ @Test
+ public void testBlobTypesPreserveBytes() throws Exception {
+ for (MysqlColType type : new MysqlColType[]
{MysqlColType.MYSQL_TYPE_TINY_BLOB,
+ MysqlColType.MYSQL_TYPE_MEDIUM_BLOB,
MysqlColType.MYSQL_TYPE_LONG_BLOB,
+ MysqlColType.MYSQL_TYPE_BLOB}) {
+ MysqlSerializer value = MysqlSerializer.newInstance();
+ byte[] bytes = new byte[] {0, (byte) 0x80, (byte) 0xff, 65};
+ value.writeLenEncodedBytes(bytes);
+ Assertions.assertEquals(new VarBinaryLiteral(bytes),
+ Literal.getLiteralByMysqlType(type, false,
value.toByteBuffer()));
+ }
+ }
+
+ @Test
+ public void testTruncatedBlobFailsBinding() {
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () ->
Literal.getLiteralByMysqlType(MysqlColType.MYSQL_TYPE_BLOB, false,
+ ByteBuffer.wrap(new byte[] {5, 65})));
+ }
+
+ @Test
+ public void testForwardedPrepareUsesParserContext() throws Exception {
+ ConnectContext context = new ConnectContext(null, true);
+ OriginStatement sql = new OriginStatement("SELECT HEX(?)", 0);
+ StmtExecutor executor = new StmtExecutor(context, sql, true);
+ StatementContext constructorContext = context.getStatementContext();
+ StatementContext parsedContext = new StatementContext(context, sql);
+ parsedContext.setPrepareStage(true);
+ LogicalPlan plan = Mockito.mock(LogicalPlan.class);
+ executor.setParsedStmt(new LogicalPlanAdapter(plan, parsedContext));
+ context.setStatementContext(parsedContext);
+ CascadesContext cascades = Mockito.mock(CascadesContext.class);
+ Mockito.when(cascades.getRewritePlan()).thenReturn(plan);
+ Mockito.when(plan.getOutput()).thenReturn(List.of());
+ try (MockedConstruction<PrepareCommandPlanner> planners =
Mockito.mockConstruction(
+ PrepareCommandPlanner.class, (planner, construction) -> {
+ Assertions.assertSame(parsedContext,
construction.arguments().get(0));
+
Mockito.when(planner.getCascadesContext()).thenReturn(cascades);
+ })) {
+ Assertions.assertFalse(constructorContext.isPrepareStage());
+
Assertions.assertTrue(executor.planPrepareStatementSlots().isEmpty());
+ Assertions.assertEquals(1, planners.constructed().size());
+ }
+ }
+
+ private static class Harness {
+ private final ConnectContext context;
+ private final MysqlChannel channel = Mockito.mock(MysqlChannel.class);
+ private final MysqlConnectProcessor processor;
+ private final PreparedStatementContext prepared;
+
+ Harness(MysqlColType... types) throws Exception {
+ this(new ConnectContext(), IntStream.range(0, types.length)
+ .mapToObj(i -> new Placeholder(new PlaceholderId(i),
types[i].getCode())).toList());
+ }
+
+ Harness(int parameterCount) throws Exception {
+ this(new ConnectContext(null, true), IntStream.range(0,
parameterCount)
+ .mapToObj(i -> new Placeholder(new
PlaceholderId(i))).toList());
+ }
+
+ private Harness(ConnectContext context, List<Placeholder>
placeholders) throws Exception {
+ this.context = context;
+ Field adapter =
ConnectContext.class.getDeclaredField("protocolAdapter");
+ adapter.setAccessible(true);
+ adapter.set(context, new MysqlProtocolAdapter(channel));
+
Mockito.when(channel.getSerializer()).thenReturn(MysqlSerializer.newInstance());
+ context.getSessionVariable().enableServeSidePreparedStatement =
true;
+ PrepareCommand command = new PrepareCommand("7", null,
placeholders,
+ new OriginStatement("select ?", 0));
+ prepared = new PreparedStatementContext(command, context, new
StatementContext(), "select ?");
+ context.addPreparedStatementContext("7", prepared);
+ processor = new MysqlConnectProcessor(context) {
+ @Override
+ protected void resolveWorkloadGroupName() {
+ }
+ };
+ }
+
+ void upload(int statement, int parameter, byte[] bytes) throws
Exception {
+ ByteBuffer packet = ByteBuffer.allocate(7 +
bytes.length).order(ByteOrder.LITTLE_ENDIAN);
+ packet.put((byte) 0x18);
+ packet.putInt(statement).putShort((short)
parameter).put(bytes).flip();
+ command(packet);
+ Assertions.assertEquals(QueryState.MysqlStateType.NOOP,
context.getState().getStateType());
+ }
+
+ void command(ByteBuffer packet) throws Exception {
+ Mockito.when(channel.fetchOnePacket()).thenReturn(packet);
+ processor.processOnce();
+ }
+
+ void execute(byte[] payload) throws Exception {
+ try (MockedConstruction<StmtExecutor> executors =
Mockito.mockConstruction(StmtExecutor.class);
+ MockedStatic<AuditLogHelper> audit =
Mockito.mockStatic(AuditLogHelper.class)) {
+ context.getState().reset();
+ processor.handleExecute(prepared.command, 7, prepared,
+
ByteBuffer.wrap(payload).order(ByteOrder.LITTLE_ENDIAN), null);
+ Assertions.assertEquals(QueryState.MysqlStateType.OK,
context.getState().getStateType(),
+ context.getState().getErrorMessage());
+ Assertions.assertEquals(1, executors.constructed().size());
+ Mockito.verify(executors.constructed().get(0)).execute();
+ }
+ }
+
+ void executeError(byte[] payload, String message) throws Exception {
+ try (MockedStatic<AuditLogHelper> audit =
Mockito.mockStatic(AuditLogHelper.class)) {
+ context.getState().reset();
+ processor.handleExecute(prepared.command, 7, prepared,
+
ByteBuffer.wrap(payload).order(ByteOrder.LITTLE_ENDIAN), null);
+ Assertions.assertEquals(QueryState.MysqlStateType.ERR,
context.getState().getStateType());
+
Assertions.assertTrue(context.getState().getErrorMessage().contains(message),
+ context.getState().getErrorMessage());
+ }
+ }
+
+ Literal bound(int parameter) {
+ return (Literal)
prepared.getStatementContext().getIdToPlaceholderRealExpr()
+
.get(prepared.command.getPlaceholders().get(parameter).getPlaceholderId());
+ }
+ }
+}
diff --git a/gensrc/thrift/FrontendService.thrift
b/gensrc/thrift/FrontendService.thrift
index e22c1be7b87..5ade99dd1e4 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -448,6 +448,8 @@ struct TMasterOpRequest {
1008: optional bool cursor_fetch_requested
// Capabilities negotiated with the original MySQL client.
1009: optional i32 mysql_capability
+ // Whether the client-facing FE can complete an empty cursor with
COM_STMT_FETCH.
+ 1010: optional bool empty_cursor_supported
}
struct TColumnDefinition {
diff --git a/regression-test/data/prepared_stmt_p0/send_long_data.out
b/regression-test/data/prepared_stmt_p0/send_long_data.out
new file mode 100644
index 00000000000..f9ec5674319
--- /dev/null
+++ b/regression-test/data/prepared_stmt_p0/send_long_data.out
@@ -0,0 +1,40 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !blob --
+a20b04215bb28f51f6871d6a94bebbc4 42 hex:00FF8041
+
+-- !after_blob --
+1
+
+-- !blob --
+079ef7341bce92c923c56623b9fc0026 42 hex:00FF8041
+
+-- !after_blob --
+1
+
+-- !blob --
+231b93478cc02c2419591997093ec8d0 42 hex:00FF8041
+
+-- !after_blob --
+1
+
+-- !character_stream --
+3ab2dbae7df7541cfbd3172de800ac0f 43 hex:
+
+-- !after_character_stream --
+2
+
+-- !inline_after_stream --
+03fdad155b7548884584c7c39b0c5cd2 44 hex:76616C7565
+
+-- !after_close --
+3
+
+-- !forwarded_blob --
+00FF8041 45
+
+-- !forwarded_blob --
+00FF8041 45
+
+-- !after_forwarding --
+4
+
diff --git a/regression-test/suites/prepared_stmt_p0/send_long_data.groovy
b/regression-test/suites/prepared_stmt_p0/send_long_data.groovy
new file mode 100644
index 00000000000..6b28a144f55
--- /dev/null
+++ b/regression-test/suites/prepared_stmt_p0/send_long_data.groovy
@@ -0,0 +1,84 @@
+// 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.
+
+import java.io.ByteArrayInputStream
+import java.io.StringReader
+import java.nio.charset.StandardCharsets
+
+suite("send_long_data") {
+ String url = getServerPrepareJdbcUrl(context.config.jdbcUrl,
"regression_test_prepared_stmt_p0") +
+
"&emulateUnsupportedPstmts=false&blobSendChunkSize=16384&socketTimeout=10000"
+ connect(context.config.jdbcUser, context.config.jdbcPassword, url) {
+ def prepared = prepareStatement "SELECT MD5(?), ?, CONCAT('hex:',
HEX(?))"
+ assertEquals(com.mysql.cj.jdbc.ServerPreparedStatement, prepared.class)
+ try {
+ // Reuse the statement with both fresh and cached parameter type
metadata.
+ ["first", "second", ""].each { text ->
+ prepared.setBlob(1, new ByteArrayInputStream(
+ (("a\u0000中文-" + text) *
2048).getBytes(StandardCharsets.UTF_8)))
+ prepared.setInt(2, 42)
+ prepared.setBinaryStream(3, new ByteArrayInputStream(new
byte[] {0, -1, -128, 65}))
+ qe_blob prepared
+ qt_after_blob "SELECT 1"
+ }
+ prepared.clearParameters()
+ prepared.setCharacterStream(1, new StringReader("中文-long-data"))
+ prepared.setInt(2, 43)
+ prepared.setBlob(3, new ByteArrayInputStream(new byte[0]))
+ qe_character_stream prepared
+ qt_after_character_stream "SELECT 2"
+
+ prepared.setString(1, "inline")
+ prepared.setInt(2, 44)
+ prepared.setString(3, "value")
+ qe_inline_after_stream prepared
+ } finally {
+ prepared.close()
+ }
+ qt_after_close "SELECT 3"
+ }
+
+ def follower = sql_return_maparray("SHOW FRONTENDS").find {
+ it.IsMaster == "false" && it.Alive == "true"
+ }
+ String forwardingUrl = url
+ if (follower != null) {
+ forwardingUrl = getServerPrepareJdbcUrl(
+ "jdbc:mysql://${follower.Host}:${follower.QueryPort}/",
"regression_test_prepared_stmt_p0", false) +
+ "&emulateUnsupportedPstmts=false&socketTimeout=10000"
+ }
+ connect(context.config.jdbcUser, context.config.jdbcPassword,
forwardingUrl) {
+ def connection = context.getConnection()
+ connection.createStatement().withCloseable { control ->
+ control.execute("SET force_forward_all_queries=false")
+ connection.prepareStatement("SELECT HEX(?), ?").withCloseable {
prepared ->
+ assertEquals(com.mysql.cj.jdbc.ServerPreparedStatement,
prepared.class)
+ control.execute("SET force_forward_all_queries=true")
+ try {
+ 2.times {
+ prepared.setBlob(1, new ByteArrayInputStream(new
byte[] {0, -1, -128, 65}))
+ prepared.setInt(2, 45)
+ qe_forwarded_blob prepared
+ }
+ } finally {
+ control.execute("SET force_forward_all_queries=false")
+ }
+ }
+ }
+ qt_after_forwarding "SELECT 4"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
