This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new f61f79ca2e3 branch-4.2: [fix](fe) Backport prepared statement protocol
fixes to branch-4.2 (#69463)
f61f79ca2e3 is described below
commit f61f79ca2e3eadecfb975ca005eaf7348300a706
Author: Calvin Kirs <[email protected]>
AuthorDate: Sat Oct 24 20:54:33 2026 +0800
branch-4.2: [fix](fe) Backport prepared statement protocol fixes to
branch-4.2 (#69463)
### What problem does this PR solve?
Related PR: #69314
Problem Summary:
- `COM_STMT_SEND_LONG_DATA` returns an unexpected response and does not
bind uploaded chunks, leaving streaming clients out of sync. Accept
chunks without replying, defer upload errors until EXECUTE, and clear
uploaded data after execution or RESET.
- Reusing cached server-prepared statements through a non-master FE can
fail because the master prepares each forwarded execution independently.
Forward complete parameter types and values, including streamed
parameters, and use the active prepare-stage context during proxy
planning.
- Empty results with `useCursorFetch=true` can hang or leave unread
packets, including when connection attributes are absent or rewritten.
Return an empty cursor, complete FETCH with `LAST_ROW_SENT`, and
preserve framing for older FEs through an optional peer capability.
- Completed EXECUTE payloads can remain attached to idle connections.
Release command payload buffers after response finalization and
forwarding finish.
Backport adaptations for branch-4.2:
- Keep protocol state in the existing
`ConnectContext`/`ConnectProcessor` architecture and preserve
branch-4.2's expansion of cached parameter types only when forwarding.
- Regenerate regression expectations with the standard test script to
retain branch-4.2's existing VARBINARY function semantics.
### Release note
Fix streamed prepared parameters, repeated prepared execution through
non-master FEs, and empty cursor fetch hangs.
### Check List (For Author)
- Test
- [x] Regression test: `prepared_stmt_p0/send_long_data` and
`prepared_stmt_p0/cursor_fetch_empty_result` passed with
`run-regression-test.sh`.
- [x] Unit Test: 91 FE tests passed with `run-fe-ut.sh`, with zero
failures, errors or skips.
- [x] Manual test: Connector/J 8.0.27, 8.0.33, 8.4.0 and 9.7.0 each
passed 428 checks against the built branch-4.2 FE and BE 4.1.3. Connect
explicitly to master and follower endpoints; verify streamed bytes,
empty streams and statement reuse; run 50 batches of three rows per
version and compare all stored values; alternate empty/nonempty cursor
results with and without connection attributes, including forwarded
execution and connection reuse.
- [x] Full FE build passed via `build.sh --fe`; Checkstyle passed with
zero violations.
- Behavior changed:
- [x] Yes. Long-data uploads have no response; forwarded parameters are
self-contained; empty cursors complete through FETCH; completed command
buffers are released.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../nereids/trees/expressions/literal/Literal.java | 6 +
.../java/org/apache/doris/qe/ConnectContext.java | 28 ++
.../java/org/apache/doris/qe/ConnectProcessor.java | 17 +-
.../java/org/apache/doris/qe/FEOpExecutor.java | 1 +
.../org/apache/doris/qe/MysqlConnectProcessor.java | 124 ++++++-
.../apache/doris/qe/PreparedStatementContext.java | 62 ++++
.../java/org/apache/doris/qe/StmtExecutor.java | 29 +-
.../doris/qe/FEOpExecutorMysqlProtocolTest.java | 2 +
.../qe/MysqlConnectProcessorCursorFetchTest.java | 24 ++
.../qe/MysqlConnectProcessorLongDataTest.java | 374 +++++++++++++++++++++
.../qe/MysqlPreparedStatementProtocolTest.java | 173 ++++++++++
gensrc/thrift/FrontendService.thrift | 2 +
.../data/prepared_stmt_p0/send_long_data.out | 40 +++
.../suites/prepared_stmt_p0/send_long_data.groovy | 84 +++++
14 files changed, 942 insertions(+), 24 deletions(-)
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 7f918305f1d..a5e1cd59ee7 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
@@ -742,6 +742,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/ConnectContext.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
index e4c231e816e..d356639ef16 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
@@ -327,6 +327,12 @@ public class ConnectContext {
// Whether the current COM_STMT_EXECUTE requested a server-side read-only
cursor.
private boolean cursorFetchRequested;
+ @Getter
+ @Setter
+ private boolean emptyCursorSupported = true;
+ private boolean mysqlResultHasRows;
+ @Getter
+ private PreparedStatementContext executingPreparedStatement;
private MysqlHandshakePacket mysqlHandshakePacket;
@@ -562,6 +568,24 @@ public class ConnectContext {
this.cursorFetchRequested = cursorFetchRequested;
}
+ public void beginPreparedExecute(PreparedStatementContext prepared) {
+ executingPreparedStatement = prepared;
+ prepared.setEmptyCursorOpen(false);
+ }
+
+ public boolean hasMysqlResultRows() {
+ return mysqlResultHasRows;
+ }
+
+ public void recordMysqlResultRow() {
+ mysqlResultHasRows = true;
+ }
+
+ public void resetMysqlResult() {
+ mysqlResultHasRows = false;
+ getMysqlChannel().reset();
+ }
+
public boolean isTxnModel() {
return txnEntry != null && txnEntry.isTxnModel();
}
@@ -1224,7 +1248,11 @@ public class ConnectContext {
statementContext = null;
loadBackendSelectionDecision = null;
loadBackendSelectionHint = null;
+ prepareExecuteBuffer = null;
+ prepareExecuteTypeCodes = null;
cursorFetchRequested = false;
+ executingPreparedStatement = null;
+ mysqlResultHasRows = false;
}
public PlSqlOperation getPlSqlOperation() {
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 eb837080f24..85cd45f1bae 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
@@ -159,11 +159,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();
}
@@ -619,6 +622,9 @@ public abstract class ConnectProcessor {
// and expected by MySQL Connector/J 9.5.0+.
if (ctx.getState().getStateType() == QueryState.MysqlStateType.EOF
&& ctx.getMysqlChannel().clientDeprecatedEOF()) {
+ if (ctx.isEmptyCursorSupported() && ctx.isCursorFetchRequested()
&& !ctx.hasMysqlResultRows()) {
+ ctx.getState().serverStatus |=
MysqlServerStatusFlag.SERVER_STATUS_CURSOR_EXISTS;
+ }
packet = new MysqlResultSetEndPacket(ctx.getState());
} else {
packet = ctx.getState().toResponsePacket();
@@ -664,6 +670,14 @@ public abstract class ConnectProcessor {
return;
}
+ PreparedStatementContext prepared =
ctx.getExecutingPreparedStatement();
+ if (prepared != null) {
+ // The master encoded a forwarded EXECUTE's final packet. Retain
its empty cursor
+ // on the client-facing FE so the following FETCH can complete
without another RPC.
+ boolean cursorOpen =
Byte.toUnsignedInt(packet.get(packet.position())) == 0xFE
+ && (packet.get(packet.position() + 3) &
MysqlServerStatusFlag.SERVER_STATUS_CURSOR_EXISTS) != 0;
+ prepared.setEmptyCursorOpen(cursorOpen);
+ }
LOG.debug("Send to mysql channel for query {}",
DebugUtil.printId(ctx.queryId));
MysqlChannel channel = ctx.getMysqlChannel();
channel.sendAndFlush(packet);
@@ -843,6 +857,7 @@ public abstract class ConnectProcessor {
// Old followers do not carry the cursor flag. Keep their existing
behavior; they must
// be upgraded to preserve cursor intent. Do not reject their ordinary
prepared statements.
context.setCursorFetchRequested(request.isSetCursorFetchRequested() &&
request.isCursorFetchRequested());
+ context.setEmptyCursorSupported(request.isSetEmptyCursorSupported() &&
request.isEmptyCursorSupported());
}
static void restoreForwardedSessionContext(ConnectContext context,
TMasterOpRequest request) {
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 42f74480acc..bcedad9321f 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
@@ -220,6 +220,7 @@ public class FEOpExecutor {
params.setPrepareExecuteBuffer(buildPrepareExecuteBuffer());
}
params.setCursorFetchRequested(ctx.isCursorFetchRequested());
+ params.setEmptyCursorSupported(true);
}
ctx.getSessionContext().getDelegatedCredential().ifPresent((DelegatedCredential
credential) -> {
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 1f46681f045..7bfcf4169d8 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
@@ -21,6 +21,7 @@ import org.apache.doris.analysis.StatementBase;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MysqlColType;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.AuthenticationException;
import org.apache.doris.common.ConnectionException;
import org.apache.doris.common.ErrorCode;
@@ -33,6 +34,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.nereids.StatementContext;
@@ -91,6 +93,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();
@@ -124,12 +147,18 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
@Override
protected void handleExecute(PrepareCommand prepareCommand, long stmtId,
PreparedStatementContext prepCtx,
ByteBuffer packetBuf, TUniqueId queryId) {
+ ctx.beginPreparedExecute(prepCtx);
int paramCount = prepareCommand.placeholderCount();
LOG.debug("execute prepared statement {}, paramCount {}", stmtId,
paramCount);
// null bitmap
String stmtStr = "";
try {
- StatementContext statementContext = prepCtx.statementContext;
+ if (prepCtx.getLongDataError() != null) {
+ ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR,
prepCtx.getLongDataError());
+ return;
+ }
+ ctx.resetMysqlResult();
+ StatementContext statementContext = prepCtx.getStatementContext();
if (!ctx.isProxy()) {
// An empty buffer still identifies a zero-parameter
COM_STMT_EXECUTE when forwarding.
ctx.setPrepareExecuteBuffer(packetBuf.duplicate());
@@ -142,7 +171,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) {
@@ -165,6 +195,23 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
}
ctx.setPrepareExecuteTypeCodes(typeCodes);
}
+ // Inline uploads with complete types: the master's fresh
PREPARE has no upload state.
+ // Without uploads, FEOpExecutor keeps expanding cached types
only when forwarding.
+ MysqlSerializer forwarded = prepCtx.hasLongData()
+ ? 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.getMysqlColType().getCode()
+ | (placeholder.isUnsigned() ?
MysqlColType.UNSIGNED_MASK : 0));
+ }
+ }
// parse param data
for (int i = 0; i < paramCount; ++i) {
PlaceholderId exprId =
prepareCommand.getPlaceholders().get(i).getPlaceholderId();
@@ -175,9 +222,39 @@ 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());
+ ctx.setPrepareExecuteTypeCodes(null);
+ }
}
ExecuteCommand executeStmt = new
ExecuteCommand(String.valueOf(stmtId), prepareCommand, statementContext);
// TODO set real origin statement
@@ -203,17 +280,16 @@ 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());
- }
- try {
- if (ctx.getSessionVariable().isEnablePreparedStmtAuditLog()) {
- auditAfterExec(stmtStr, executor.getParsedStmt(),
executor.getQueryStatisticsForAuditLog(), true);
- } else {
- // When audit log is disabled for prepared statements, still
update QPS metrics.
- AuditLogHelper.updateMetrics(ctx);
- }
} finally {
+ prepCtx.clearLongData();
prepCtx.statementContext.clearExternalScanTasks();
}
+ 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.
+ AuditLogHelper.updateMetrics(ctx);
+ }
}
// process COM_EXECUTE, parse binary row data
@@ -245,6 +321,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
@@ -289,6 +382,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 d34f802ddb1..c09d7fa3217 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;
}
@@ -74,4 +90,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 cc20e456b84..6c1ebaedc68 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
@@ -1294,7 +1294,7 @@ public class StmtExecutor {
isSend = true;
}
for (ByteBuffer row : batch.getBatch().getRows()) {
- channel.sendOnePacket(row);
+ sendMysqlResultRow(row);
}
context.updateReturnRows(batch.getBatch().getRows().size());
}
@@ -1335,7 +1335,7 @@ public class StmtExecutor {
if (context.getConnectType() == ConnectType.MYSQL) {
// Every time set no send flag and clean all data in buffer
- context.getMysqlChannel().reset();
+ context.resetMysqlResult();
}
Queriable queryStmt = (Queriable) parsedStmt;
@@ -1526,7 +1526,7 @@ public class StmtExecutor {
isSendFields = true;
}
for (ByteBuffer row : batch.getBatch().getRows()) {
- channel.sendOnePacket(row);
+ sendMysqlResultRow(row);
}
profile.getSummaryProfile().freshWriteResultConsumeTime();
context.updateReturnRows(batch.getBatch().getRows().size());
@@ -1880,10 +1880,8 @@ public class StmtExecutor {
serializer.reset();
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.
+ } else if (!context.isEmptyCursorSupported() &&
connectorJConsumesCursorMetadataTerminator()) {
+ // Preserve framing for older client-facing FEs during rolling
upgrades.
serializer.reset();
new
MysqlResultSetEndPacket(context.getState()).writeTo(serializer);
channel.sendOnePacket(serializer.toByteBuffer());
@@ -1926,6 +1924,11 @@ public class StmtExecutor {
}
}
+ private void sendMysqlResultRow(ByteBuffer row) throws IOException {
+ context.recordMysqlResultRow();
+ context.getMysqlChannel().sendOnePacket(row);
+ }
+
protected void sendTextResultRow(ResultSet resultSet) throws IOException {
for (List<String> row : resultSet.getResultRows()) {
serializer.reset();
@@ -1936,7 +1939,7 @@ public class StmtExecutor {
serializer.writeLenEncodedString(item);
}
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ sendMysqlResultRow(serializer.toByteBuffer());
}
}
@@ -2003,7 +2006,7 @@ public class StmtExecutor {
}
}
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ sendMysqlResultRow(serializer.toByteBuffer());
}
}
@@ -2049,7 +2052,7 @@ public class StmtExecutor {
for (String item : result.split("\n")) {
serializer.reset();
serializer.writeLenEncodedString(item);
-
context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ sendMysqlResultRow(serializer.toByteBuffer());
}
} else if (context.getConnectType() == ConnectType.ARROW_FLIGHT_SQL) {
context.getFlightSqlChannel()
@@ -2071,7 +2074,7 @@ public class StmtExecutor {
for (String item : result.split("\n")) {
serializer.reset();
serializer.writeLenEncodedString(item);
-
context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ sendMysqlResultRow(serializer.toByteBuffer());
}
} else if (context.getConnectType() == ConnectType.ARROW_FLIGHT_SQL) {
context.getFlightSqlChannel()
@@ -2130,7 +2133,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/qe/FEOpExecutorMysqlProtocolTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorMysqlProtocolTest.java
index de0c3e2ee8e..28129a95f61 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
@@ -62,6 +62,7 @@ public class FEOpExecutorMysqlProtocolTest {
Assertions.assertTrue(request.isClientDeprecatedEOF());
Assertions.assertEquals(context.getCapability().getFlags(),
request.getMysqlCapability());
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})) {
@@ -73,6 +74,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..60b0b248b26
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
@@ -0,0 +1,374 @@
+// 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.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.apache.doris.thrift.TNetworkAddress;
+
+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.Method;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+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());
+ Assertions.assertNull(h.context.getPrepareExecuteTypeCodes());
+
+ 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.assertArrayEquals(cached.toArray(),
h.context.getPrepareExecuteBuffer().array());
+ Assertions.assertNotNull(h.context.getPrepareExecuteTypeCodes());
+ 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 {
+ Method build =
FEOpExecutor.class.getDeclaredMethod("buildPrepareExecuteBuffer");
+ build.setAccessible(true);
+ FEOpExecutor forwarder = new FEOpExecutor(new
TNetworkAddress("127.0.0.1", 9010),
+ new OriginStatement("select ?", 0), follower.context, true);
+ ByteBuffer forwarded = ((ByteBuffer)
build.invoke(forwarder)).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.assertThrows(java.util.NoSuchElementException.class,
placeholder::getMysqlColType);
+ }
+ 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).getMysqlColType(),
+
master.prepared.command.getPlaceholders().get(i).getMysqlColType());
+
Assertions.assertEquals(follower.prepared.command.getPlaceholders().get(i).isUnsigned(),
+
master.prepared.command.getPlaceholders().get(i).isUnsigned());
+ }
+ }
+
+ @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(Collections.emptyList());
+ 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()))
+ .collect(Collectors.toList()));
+ }
+
+ Harness(int parameterCount) throws Exception {
+ this(new ConnectContext(null, true), IntStream.range(0,
parameterCount)
+ .mapToObj(i -> new Placeholder(new
PlaceholderId(i))).collect(Collectors.toList()));
+ }
+
+ private Harness(ConnectContext context, List<Placeholder>
placeholders) throws Exception {
+ this.context = context;
+ context.mysqlChannel = 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/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlPreparedStatementProtocolTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlPreparedStatementProtocolTest.java
new file mode 100644
index 00000000000..fee1b1b00d5
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlPreparedStatementProtocolTest.java
@@ -0,0 +1,173 @@
+// 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.Column;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.mysql.DummyMysqlChannel;
+import org.apache.doris.mysql.MysqlCommand;
+import org.apache.doris.mysql.MysqlSerializer;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.qe.CommonResultSet.CommonResultSetMetaData;
+import org.apache.doris.thrift.TMasterOpRequest;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class MysqlPreparedStatementProtocolTest {
+ @Test
+ public void testEmptyCursorDoesNotDependOnClientAttributes() throws
Exception {
+ for (Map<String, String> attributes :
Arrays.asList(Collections.<String, String>emptyMap(),
+ ImmutableMap.of("_client_name", "proxy", "_client_version",
"0"),
+ ImmutableMap.of("_client_name", "MySQL Connector/J",
"_client_version", "8.0.27"),
+ ImmutableMap.of("_client_name", "MySQL Connector/J",
"_client_version", "9.7.0"))) {
+ ConnectContext context = context();
+ context.setConnectAttributes(attributes);
+ RecordingChannel channel = (RecordingChannel)
context.getMysqlChannel();
+ PreparedStatementContext prepared = prepared(context);
+ StmtExecutor executor = new StmtExecutor(context, new
OriginStatement("select 1", 0), true);
+ executor.sendResultSet(result(Collections.emptyList()));
+ MysqlConnectProcessor processor = new
MysqlConnectProcessor(context);
+ processor.finalizeCommand();
+
+ // Column count, one column definition, and one unambiguous
cursor-opening packet.
+ Assertions.assertEquals(3, channel.packets.size());
+ Assertions.assertEquals(0x40,
Byte.toUnsignedInt(channel.last().get(3)) & 0xC0);
+ Assertions.assertTrue(prepared.isEmptyCursorOpen());
+ context.clear();
+ Assertions.assertNull(context.getExecutingPreparedStatement());
+ Assertions.assertTrue(prepared.isEmptyCursorOpen());
+
+ context.getState().reset();
+ processor.handleStmtFetch(7);
+ processor.finalizeCommand();
+ Assertions.assertEquals(0x80,
Byte.toUnsignedInt(channel.last().get(3)) & 0xC0);
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+ }
+ }
+
+ @Test
+ public void testBinaryNullRowIsNotAnEmptyCursor() throws Exception {
+ ConnectContext context = context();
+ PreparedStatementContext prepared = prepared(context);
+ StmtExecutor executor = new StmtExecutor(context, new
OriginStatement("select null", 0), true);
+
executor.sendBinaryResultRow(result(Collections.singletonList(Collections.singletonList(null))));
+ Assertions.assertArrayEquals(new byte[] {0, 4},
+ ((RecordingChannel) context.getMysqlChannel()).last().array());
+ context.getState().setEof();
+ new MysqlConnectProcessor(context).finalizeCommand();
+ Assertions.assertTrue(context.hasMysqlResultRows());
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+
+ context.resetMysqlResult();
+ Assertions.assertFalse(context.hasMysqlResultRows());
+ context.getState().reset();
+ executor.sendResultSet(result(Collections.emptyList()));
+ new MysqlConnectProcessor(context).finalizeCommand();
+ Assertions.assertTrue(prepared.isEmptyCursorOpen());
+ }
+
+ @Test
+ public void testForwardedCursorIsRetainedOnClientFacingFrontend() throws
Exception {
+ ConnectContext context = context();
+ PreparedStatementContext prepared = prepared(context);
+ 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}));
+ MysqlConnectProcessor processor = new MysqlConnectProcessor(context);
+ processor.executor = executor;
+ processor.finalizeCommand();
+ Assertions.assertTrue(prepared.isEmptyCursorOpen());
+ Mockito.verify(executor).sendProxyQueryResult();
+ context.beginPreparedExecute(prepared);
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+ }
+
+ @Test
+ public void testOldFrontendKeepsLegacyMetadataFraming() throws Exception {
+ ConnectContext context = context();
+ context.setConnectAttributes(ImmutableMap.of(
+ "_client_name", "MySQL Connector/J", "_client_version",
"8.0.27"));
+ TMasterOpRequest request = new TMasterOpRequest();
+ request.setClientDeprecatedEOF(true);
+ request.setCursorFetchRequested(true);
+ ConnectProcessor.restoreForwardedMysqlContext(context, request);
+ Assertions.assertFalse(context.isEmptyCursorSupported());
+ PreparedStatementContext prepared = prepared(context);
+ new StmtExecutor(context, new OriginStatement("select 1", 0), true)
+ .sendResultSet(result(Collections.emptyList()));
+ new MysqlConnectProcessor(context).finalizeCommand();
+ Assertions.assertEquals(4, ((RecordingChannel)
context.getMysqlChannel()).packets.size());
+ Assertions.assertFalse(prepared.isEmptyCursorOpen());
+ }
+
+ private ConnectContext context() {
+ ConnectContext context = new ConnectContext();
+ context.mysqlChannel = new RecordingChannel();
+ context.getMysqlChannel().setClientDeprecatedEOF();
+ context.setCursorFetchRequested(true);
+ context.setCommand(MysqlCommand.COM_STMT_EXECUTE);
+ return context;
+ }
+
+ private PreparedStatementContext prepared(ConnectContext context) throws
Exception {
+ PreparedStatementContext prepared = new PreparedStatementContext(null,
context, new StatementContext(), "7");
+ context.addPreparedStatementContext("7", prepared);
+ context.beginPreparedExecute(prepared);
+ return prepared;
+ }
+
+ private CommonResultSet result(List<List<String>> rows) {
+ return new CommonResultSet(new CommonResultSetMetaData(
+ Collections.singletonList(new Column("v",
PrimitiveType.INT))), rows);
+ }
+
+ private static class RecordingChannel extends DummyMysqlChannel {
+ private final List<ByteBuffer> packets = new ArrayList<>();
+
+ private RecordingChannel() {
+ serializer = MysqlSerializer.newInstance();
+ }
+
+ @Override
+ public void sendOnePacket(ByteBuffer packet) {
+ byte[] copy = new byte[packet.remaining()];
+ packet.duplicate().get(copy);
+ packets.add(ByteBuffer.wrap(copy));
+ }
+
+ @Override
+ public void sendAndFlush(ByteBuffer packet) {
+ sendOnePacket(packet);
+ }
+
+ private ByteBuffer last() {
+ return packets.get(packets.size() - 1);
+ }
+ }
+}
diff --git a/gensrc/thrift/FrontendService.thrift
b/gensrc/thrift/FrontendService.thrift
index c5112c1d2dc..87027b02a4c 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -442,6 +442,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 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..a84de53f7df
--- /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:3030464638303431
+
+-- !after_blob --
+1
+
+-- !blob --
+079ef7341bce92c923c56623b9fc0026 42 hex:3030464638303431
+
+-- !after_blob --
+1
+
+-- !blob --
+231b93478cc02c2419591997093ec8d0 42 hex:3030464638303431
+
+-- !after_blob --
+1
+
+-- !character_stream --
+3ab2dbae7df7541cfbd3172de800ac0f 43 hex:
+
+-- !after_character_stream --
+2
+
+-- !inline_after_stream --
+03fdad155b7548884584c7c39b0c5cd2 44 hex:76616C7565
+
+-- !after_close --
+3
+
+-- !forwarded_blob --
+3030464638303431 45
+
+-- !forwarded_blob --
+3030464638303431 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]
