Skip to content

Commit

Permalink
Support COM_CHANGE_USER and other mysql command.
Browse files Browse the repository at this point in the history
  • Loading branch information
@Jibing-Li
Jibing-Li committed Sep 18, 2024
1 parent 5c9ac9a commit ac4137d
Show file tree
Hide file tree
Showing 5 changed files with 121 additions and 0 deletions.
2 changes: 2 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlCommand.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@ public enum MysqlCommand {
COM_STMT_SEND_LONG_DATA("COM_STMT_SEND_LONG_DATA", 24),
COM_STMT_CLOSE("COM_STMT_CLOSE", 25),
COM_STMT_RESET("COM_STMT_RESET", 26),
COM_SET_OPTION("COM_RESET_CONNECTION", 27),
COM_STMT_FETCH("COM_RESET_CONNECTION", 28),
COM_DAEMON("COM_DAEMON", 29),
COM_RESET_CONNECTION("COM_RESET_CONNECTION", 31);

Expand Down
1 change: 1 addition & 0 deletions fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@ public static boolean negotiate(ConnectContext context) throws IOException {
serializer.reset();
MysqlHandshakePacket handshakePacket = new MysqlHandshakePacket(context.getConnectionId());
handshakePacket.writeTo(serializer);
context.setMysqlHandshakePacket(handshakePacket);
try {
channel.sendAndFlush(serializer.toByteBuffer());
} catch (IOException e) {
Expand Down
11 changes: 11 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
import org.apache.doris.mysql.MysqlCapability;
import org.apache.doris.mysql.MysqlChannel;
import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.mysql.MysqlHandshakePacket;
import org.apache.doris.mysql.MysqlSslContext;
import org.apache.doris.mysql.ProxyMysqlChannel;
import org.apache.doris.mysql.privilege.PrivPredicate;
Expand Down Expand Up @@ -249,6 +250,8 @@ public enum ConnectType {
// it's default thread-safe
private boolean isProxy = false;

private MysqlHandshakePacket mysqlHandshakePacket;

public void setUserQueryTimeout(int queryTimeout) {
if (queryTimeout > 0) {
sessionVariable.setQueryTimeoutS(queryTimeout);
Expand Down Expand Up @@ -1368,4 +1371,12 @@ public void setUserVars(Map<String, LiteralExpr> userVars) {
public boolean isProxy() {
return isProxy;
}

public void setMysqlHandshakePacket(MysqlHandshakePacket mysqlHandshakePacket) {
this.mysqlHandshakePacket = mysqlHandshakePacket;
}

public byte[] getAuthPluginData() {
return mysqlHandshakePacket == null ? null : mysqlHandshakePacket.getAuthPluginData();
}
}
5 changes: 5 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,11 @@ protected void handlePing() {
ctx.getState().setOk();
}

// Do nothing for now.
protected void handleStatistics() {
ctx.getState().setOk();
}

protected void handleStmtReset() {
ctx.getState().setOk();
}
Expand Down
102 changes: 102 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,22 @@
package org.apache.doris.qe;

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.cloud.catalog.CloudEnv;
import org.apache.doris.common.AuthenticationException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ConnectionException;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.mysql.MysqlChannel;
import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.mysql.MysqlProto;
import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.trees.expressions.Placeholder;
Expand All @@ -33,6 +42,8 @@
import org.apache.doris.nereids.trees.plans.commands.ExecuteCommand;
import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

Expand Down Expand Up @@ -231,6 +242,12 @@ private void dispatch() throws IOException {
// process COM_PING statement, do nothing, just return one OK packet.
handlePing();
break;
case COM_STATISTICS:
handleStatistics();
break;
case COM_CHANGE_USER:
handleChangeUser();
break;
case COM_STMT_RESET:
handleStmtReset();
break;
Expand All @@ -249,6 +266,91 @@ private void handleFieldList() throws ConnectionException {
handleFieldList(tableName);
}

private void handleChangeUser() {
// Random bytes generated when creating connection.
byte[] authPluginData = getConnectContext().getAuthPluginData();
String userName = new String(MysqlProto.readNulTerminateString(packetBuf));
int passwordLen = MysqlProto.readInt1(packetBuf);
byte[] password = MysqlProto.readFixedString(packetBuf, passwordLen);
String db = new String(MysqlProto.readNulTerminateString(packetBuf));

// For safety, not allowed to change to root or admin.
if (Auth.ROOT_USER.equals(userName) || Auth.ADMIN_USER.equals(userName)) {
ctx.getState().setError(ErrorCode.ERR_ACCESS_DENIED_ERROR, "Change to root or admin is forbidden");
return;
}

// Check password.
List<UserIdentity> currentUserIdentity = Lists.newArrayList();
try {
Env.getCurrentEnv().getAuth()
.checkPassword(userName, ctx.remoteIP, password, authPluginData, currentUserIdentity);
} catch (AuthenticationException e) {
ctx.getState().setError(ErrorCode.ERR_ACCESS_DENIED_ERROR, "Authentication failed.");
return;
}
ctx.setCurrentUserIdentity(currentUserIdentity.get(0));
ctx.setQualifiedUser(userName);

// Change default db if set.
if (Strings.isNullOrEmpty(db)) {
ctx.changeDefaultCatalog(InternalCatalog.INTERNAL_CATALOG_NAME);
} else {
String catalogName = null;
String dbName = null;
String[] dbNames = db.split("\\.");
if (dbNames.length == 1) {
dbName = db;
} else if (dbNames.length == 2) {
catalogName = dbNames[0];
dbName = dbNames[1];
} else if (dbNames.length > 2) {
ctx.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "Only one dot can be in the name: " + db);
return;
}

// mysql -d
if (Config.isCloudMode()) {
try {
dbName = ((CloudEnv) Env.getCurrentEnv()).analyzeCloudCluster(dbName, ctx);
} catch (DdlException e) {
ctx.getState().setError(e.getMysqlErrorCode(), e.getMessage());
return;
}

if (dbName == null || dbName.isEmpty()) {
ctx.getState().setOk();
return;
}
}

String dbFullName = dbName;

// check catalog and db exists
if (catalogName != null) {
CatalogIf catalogIf = ctx.getEnv().getCatalogMgr().getCatalog(catalogName);
if (catalogIf == null) {
ctx.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "No match catalog in doris: " + db);
return;
}
if (catalogIf.getDbNullable(dbFullName) == null) {
ctx.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "No match database in doris: " + db);
return;
}
}
try {
if (catalogName != null) {
ctx.getEnv().changeCatalog(ctx, catalogName);
}
Env.getCurrentEnv().changeDb(ctx, dbFullName);
} catch (DdlException e) {
ctx.getState().setError(e.getMysqlErrorCode(), e.getMessage());
return;
}
}
ctx.getState().setOk();
}

// Process a MySQL request
public void processOnce() throws IOException {
// set status of query to OK.
Expand Down

0 comments on commit ac4137d

Please sign in to comment.