[feat][dingo-mysql] Support: 1.create user require ssl 2. interactive…

…_timeout 3. fix bug: timeZone

dingo-calcite/src/main/codegen/config.fmpp

@@ -36,6 +36,7 @@ data: {
       "io.dingodb.calcite.grammar.ddl.SqlCommit"
       "io.dingodb.calcite.grammar.ddl.SqlRollback"
       "io.dingodb.calcite.grammar.ddl.SqlAlterAddIndex"
+      "io.dingodb.calcite.grammar.ddl.SqlAlterUser"
       "io.dingodb.calcite.grammar.ddl.SqlCreateUser"
       "io.dingodb.calcite.grammar.ddl.SqlCreateIndex"
       "io.dingodb.calcite.grammar.ddl.SqlDropUser"
@@ -118,6 +119,10 @@ data: {
       "SqlDropIndex"
     ]
 
+    alterStatementParserMethods: [
+       "SqlAlterUser"
+    ]
+
     # List of files in @includes directory that have parser method
     # implementations for parsing custom SQL statements, literals or types
     # given as part of "statementParserMethods", "literalParserMethods" or

dingo-calcite/src/main/codegen/includes/parserImpls.ftl

@@ -259,15 +259,17 @@ SqlCreate SqlCreateUser(Span s, boolean replace) :
     String host = "%";
     SqlNode create = null;
     Boolean ifNotExists = false;
+    String requireSsl = null;
 }
 {
     <USER> ifNotExists = IfNotExistsOpt()
     ( <QUOTED_STRING> | <IDENTIFIER> )
      { user = token.image; }
     [ <AT_SPLIT> (<QUOTED_STRING> | <IDENTIFIER>) { host = token.image; }  ]
     <IDENTIFIED> <BY>  <QUOTED_STRING> { password = token.image; }
+    [ <REQUIRE> (  <SSL> { requireSsl = "SSL"; } | <NONE> { requireSsl  = "NONE"; }) ]
     {
-       return new SqlCreateUser(user, password, host, s.end(this), replace, ifNotExists);
+       return new SqlCreateUser(user, password, host, s.end(this), replace, ifNotExists, requireSsl);
     }
 }
 
@@ -1056,3 +1058,21 @@ SqlExecute SqlExecute(): {
       return new SqlExecute(s.end(this), statementName, paramList);
    }
 }
+
+SqlAlterUser SqlAlterUser(Span s, String scope): {
+    final String user;
+    String password = null;
+    String host = "%";
+    SqlNode create = null;
+    String requireSsl = null;
+} {
+   <USER>
+   ( <QUOTED_STRING> | <IDENTIFIER> )
+     { s = span(); user = token.image; }
+    [ <AT_SPLIT> (<QUOTED_STRING> | <IDENTIFIER>) { host = token.image; }  ]
+    [ <IDENTIFIED> <BY>  <QUOTED_STRING> { password = token.image; } ]
+    [ <REQUIRE> (  <SSL> { requireSsl = "SSL"; } | <NONE> { requireSsl  = "NONE"; }) ]
+    {
+      return new SqlAlterUser(user, password, host, requireSsl, s.end(this));
+    }
+}

dingo-calcite/src/main/codegen/templates/Parser.jj

@@ -4221,6 +4221,7 @@ SqlSetOption SqlSetOption(Span s, String scope) :
               }
         ]
         [ <GLOBAL> { scope = "SYSTEM"; } ]
+        [ <SESSION> { scope = "SESSION"; } ]
         [ <EXECUTOR> { scope = "EXECUTOR"; } ]
         [ <AT_SPLIT> { scope = "USER"; } ]
         name = CompoundIdentifier()
@@ -4274,7 +4275,7 @@ SqlSetOption SqlSetOption(Span s, String scope) :
 SqlAlter SqlAlter() :
 {
     final Span s;
-    final String scope;
+    String scope = null;
     final SqlIdentifier id;
     final String index;
     boolean isUnique = false;
@@ -4284,7 +4285,7 @@ SqlAlter SqlAlter() :
 }
 {
     <ALTER> { s = span(); }
-    scope = Scope()
+    [ scope = Scope() ]
     [
     (
 <#-- additional literal parser methods are included here -->
@@ -8102,6 +8103,7 @@ SqlPostfixOperator PostfixRowOperator() :
 |   < RELOAD: "RELOAD" >
 |   < REPEATABLE: "REPEATABLE" >
 |   < REPLACE: "REPLACE" >
+|   < REQUIRE: "REQUIRE" >
 |   < RESET: "RESET" >
 |   < RESPECT: "RESPECT" >
 |   < RESTART: "RESTART" >
@@ -8225,6 +8227,7 @@ SqlPostfixOperator PostfixRowOperator() :
 |   < SQL_VARBINARY: "SQL_VARBINARY" >
 |   < SQL_VARCHAR: "SQL_VARCHAR" >
 |   < SQRT: "SQRT" >
+|   < SSL: "SSL" >
 |   < START: "START" >
 |   < STATE: "STATE" >
 |   < STATEMENT: "STATEMENT" >

dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java

@@ -18,6 +18,7 @@
 
 import io.dingodb.calcite.grammar.ddl.DingoSqlCreateTable;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddIndex;
+import io.dingodb.calcite.grammar.ddl.SqlAlterUser;
 import io.dingodb.calcite.grammar.ddl.SqlCommit;
 import io.dingodb.calcite.grammar.ddl.SqlCreateIndex;
 import io.dingodb.calcite.grammar.ddl.SqlCreateUser;
@@ -426,6 +427,7 @@ public void execute(@NonNull SqlCreateUser sqlCreateUser, CalcitePrepare.Context
             throw new RuntimeException("user is exists");
         } else {
             userDefinition.setPlugin("mysql_native_password");
+            userDefinition.setRequireSsl(sqlCreateUser.requireSsl);
             String digestPwd = AlgorithmPlugin.digestAlgorithm(sqlCreateUser.password, userDefinition.getPlugin());
             userDefinition.setPassword(digestPwd);
             userService.createUser(userDefinition);
@@ -453,7 +455,7 @@ public void execute(@NonNull SqlSetPassword sqlSetPassword, CalcitePrepare.Conte
             .build();
         if (userService.existsUser(userDefinition)) {
             userDefinition.setPassword(sqlSetPassword.password);
-            userService.setPassword(userDefinition);
+            userService.updateUser(userDefinition);
         } else {
             throw new RuntimeException("user is not exist");
         }
@@ -488,6 +490,20 @@ public void execute(@NonNull SqlDropIndex sqlDropIndex, CalcitePrepare.Context c
         schema.dropIndex(tableName, sqlDropIndex.index);
     }
 
+    public void execute(@NonNull SqlAlterUser sqlAlterUser, CalcitePrepare.Context context) {
+        UserDefinition userDefinition = UserDefinition.builder()
+            .user(sqlAlterUser.user)
+            .host(getRealAddress(sqlAlterUser.host))
+            .build();
+        if (userService.existsUser(userDefinition)) {
+            userDefinition.setPassword(sqlAlterUser.password);
+            userDefinition.setRequireSsl(sqlAlterUser.requireSsl);
+            userService.updateUser(userDefinition);
+        } else {
+            throw new RuntimeException("user is not exist");
+        }
+    }
+
     public void execute(@NonNull SqlSetOption sqlSetOption, CalcitePrepare.Context context) {
         log.info("sql set option");
     }

dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterUser.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * Licensed 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 io.dingodb.calcite.grammar.ddl;
+
+import org.apache.calcite.sql.SqlAlter;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+import java.util.List;
+
+public class SqlAlterUser extends SqlAlter {
+
+    public final String password;
+    public final String user;
+    public String host;
+    public String requireSsl;
+
+    private static final SqlOperator OPERATOR =
+        new SqlSpecialOperator("ALTER USER", SqlKind.OTHER_DDL);
+
+    public SqlAlterUser(String user, String password, String host, String requireSsl, SqlParserPos pos) {
+        super(pos);
+        this.password = password.contains("'") ? password.replace("'", "") : password;
+        this.user = user.contains("'") ? user.replace("'", "") : user;
+        this.host = host == null ? "%" : host.contains("'") ? host.replace("'", "") : host;
+        this.requireSsl = requireSsl;
+    }
+
+    @Override
+    public void unparseAlterOperation(SqlWriter writer, int leftPrec, int rightPrec) {
+
+    }
+
+    @Override
+    public SqlOperator getOperator() {
+        return OPERATOR;
+    }
+
+    @Override
+    public List<SqlNode> getOperandList() {
+        return null;
+    }
+}

dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlCreateUser.java

@@ -23,23 +23,31 @@
 import org.apache.calcite.sql.SqlSpecialOperator;
 import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
 
 public class SqlCreateUser extends SqlCreate {
     public final String password;
     public final String user;
     public String host;
+    public String requireSsl;
 
     private static final SqlOperator OPERATOR =
         new SqlSpecialOperator("CREATE USER", SqlKind.OTHER_DDL);
 
-    public SqlCreateUser(String user, String password, String host,
-                              SqlParserPos pos, boolean replace, boolean ifNotExists) {
+    public SqlCreateUser(String user,
+                         String password,
+                         String host,
+                         SqlParserPos pos,
+                         boolean replace,
+                         boolean ifNotExists,
+                         String requireSsl) {
         super(OPERATOR, pos, replace, ifNotExists);
         this.password = password.contains("'") ? password.replace("'", "") : password;
         this.user = user.contains("'") ? user.replace("'", "") : user;
         this.host = host == null ? "%" : host.contains("'") ? host.replace("'", "") : host;
+        this.requireSsl = requireSsl;
     }
 
     @Override
@@ -54,7 +62,11 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
         writer.keyword(user);
         writer.keyword("@");
         writer.keyword(host);
-        writer.keyword(" identified by ");
+        writer.keyword(" IDENTIFIED BY ");
         writer.keyword(password);
+        if (StringUtils.isNotBlank(requireSsl)) {
+            writer.keyword(" REQUIRE ");
+            writer.keyword(requireSsl);
+        }
     }
 }

dingo-cli/driver-mysql/src/main/java/io/dingodb/driver/mysql/IdleTimeWatcher.java

@@ -32,9 +32,18 @@ public void update(Observable observable, Object arg) {
         SessionVariableChange sessionVariable = (SessionVariableChange) arg;
         if ("wait_timeout".equalsIgnoreCase(sessionVariable.getName())) {
             MysqlConnection connection = MysqlNettyServer.connections.get(sessionVariable.getId());
-            connection.mysqlIdleStateHandler.setIdleTimeout(Long.parseLong(sessionVariable.getValue()),
-                TimeUnit.SECONDS);
-            log.info("update connection idle time:" + sessionVariable);
+            if (connection != null && !connection.authPacket.interActive) {
+                connection.mysqlIdleStateHandler.setIdleTimeout(Long.parseLong(sessionVariable.getValue()),
+                    TimeUnit.SECONDS);
+                log.info("update connection idle time:" + sessionVariable);
+            }
+        } else if ("interactive_timeout".equalsIgnoreCase(sessionVariable.getName())) {
+            MysqlConnection connection = MysqlNettyServer.connections.get(sessionVariable.getId());
+            if (connection != null && connection.authPacket.interActive) {
+                connection.mysqlIdleStateHandler.setIdleTimeout(Long.parseLong(sessionVariable.getValue()),
+                    TimeUnit.SECONDS);
+                log.info("update interactive connection idle time:" + sessionVariable);
+            }
         }
 
     }

dingo-cli/driver-mysql/src/main/java/io/dingodb/driver/mysql/command/MysqlResponseHandler.java

@@ -124,7 +124,7 @@ private static void handlerRowPacket(ResultSet resultSet, AtomicLong packetId, M
             ResultSetRowPacket resultSetRowPacket = new ResultSetRowPacket();
             resultSetRowPacket.packetId = (byte) packetId.getAndIncrement();
             for (int i = 1; i <= columnCount; i ++) {
-                resultSetRowPacket.addColumnValue(resultSet.getString(i));
+                resultSetRowPacket.addColumnValue(resultSet.getObject(i));
             }
             resultSetRowPacket.write(buffer);
             int writerIndex = buffer.writerIndex();

dingo-cli/driver-mysql/src/main/java/io/dingodb/driver/mysql/netty/HandshakeHandler.java

@@ -141,6 +141,18 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
                     //mysql protocol packet auto increment based by 0;
                     packetId.incrementAndGet();
                     if (isUserExists && validator(dbPwd, fullSeed, authPacket.password)) {
+                        if (StringUtils.isNotEmpty(userDefinition.getRequireSsl())) {
+                            if (ctx.channel().pipeline().get("tls") == null) {
+                                String error =
+                                    String.format(ErrorCode.ER_ACCESS_DENIED_ERROR.message, user, ip, "YES");
+                                MysqlResponseHandler.responseError(packetId,
+                                    mysqlConnection.channel, ErrorCode.ER_ACCESS_DENIED_ERROR, error);
+                                if (mysqlConnection.channel.isActive()) {
+                                    mysqlConnection.channel.close();
+                                }
+                                return;
+                            }
+                        }
                         OKPacket okPacket = new OKPacket();
                         okPacket.capabilities = MysqlServer.getServerCapabilities();
                         okPacket.affectedRows = 0;

dingo-cli/driver-mysql/src/main/java/io/dingodb/driver/mysql/packet/AuthPacket.java

@@ -37,10 +37,15 @@ public class AuthPacket extends MysqlPacket {
 
     public boolean isSSL;
 
+    public boolean interActive;
+
     @Override
     public void read(byte[] data) {
         MysqlMessage mm = new MysqlMessage(data);
         clientFlags = mm.readUB2();
+        if ((clientFlags & CapabilityFlags.CLIENT_INTERACTIVE.getCode()) != 0) {
+            interActive = true;
+        }
         extendClientFlags = mm.readUB2();
         maxPacketSize = mm.readUB4();
         charsetIndex = (mm.read() & 0xff);

dingo-cli/driver-mysql/src/main/java/io/dingodb/driver/mysql/packet/ResultSetRowPacket.java

@@ -115,10 +115,10 @@ public String toString() {
                 + "}\n";
     }
 
-    public void addColumnValue(String val) {
-        if (StringUtils.isBlank(val)) {
+    public void addColumnValue(Object val) {
+        if (val == null) {
             val = "";
         }
-        values.add(val);
+        values.add(val.toString());
     }
 }

dingo-common/src/main/java/io/dingodb/common/privilege/UserDefinition.java

@@ -29,15 +29,17 @@
 public class UserDefinition extends PrivilegeDefinition {
     private String plugin;
     private String password;
+    private String requireSsl;
 
     Boolean[] privileges;
 
     @Builder(toBuilder = true)
     public UserDefinition(String user, String host, String plugin,
-                          String password) {
+                          String password, String requireSsl) {
         super(user, host);
         this.plugin = plugin;
         this.password = password;
+        this.requireSsl = requireSsl;
     }
 
     public String getKey() {