feat(open-mysql-db): pandas support

extensions/open-mysql-db/python-testcases/main.py

@@ -0,0 +1,33 @@
+import pandas as pd
+from sqlalchemy import create_engine
+
+if __name__ == '__main__':
+    # Create a Pandas DataFrame (replace this with your actual data)
+    data = {'id': [1, 2, 3],
+            'name': ['Alice', 'Bob', 'Charlie'],
+            'age': [25, 30, 35],
+            'score': [1.1, 2.2, 3.3],
+            'ts': [pd.Timestamp.utcnow().timestamp(), pd.Timestamp.utcnow().timestamp(),
+                   pd.Timestamp.utcnow().timestamp()],
+            'dt': [pd.to_datetime('20240101', format='%Y%m%d'), pd.to_datetime('20240201', format='%Y%m%d'),
+                   pd.to_datetime('20240301', format='%Y%m%d')],
+            }
+    df = pd.DataFrame(data)
+
+    # Create a MySQL database engine using SQLAlchemy
+    engine = create_engine('mysql+pymysql://root:[email protected]:3307/demo_db')
+
+    # Replace 'username', 'password', 'host', and 'db_name' with your actual database credentials
+
+    # Define the name of the table in the database where you want to write the data
+    table_name = 'demo_table1'
+
+    # Write the DataFrame 'df' into the MySQL table
+    df.to_sql(table_name, engine, if_exists='replace', index=False)
+
+    # 'if_exists' parameter options:
+    # - 'fail': If the table already exists, an error will be raised.
+    # - 'replace': If the table already exists, it will be replaced.
+    # - 'append': If the table already exists, data will be appended to it.
+
+    print("Data written to MySQL table successfully!")

extensions/open-mysql-db/src/main/java/cn/paxos/mysql/MySqlListener.java

@@ -217,6 +217,9 @@ private void handleQuery(
         && !queryStringWithoutComment.startsWith("set @@execute_mode=")) {
       // ignore SET command
       ctx.writeAndFlush(OkResponse.builder().sequenceId(query.getSequenceId() + 1).build());
+    } else if (queryStringWithoutComment.equalsIgnoreCase("COMMIT")) {
+      // ignore COMMIT command
+      ctx.writeAndFlush(OkResponse.builder().sequenceId(query.getSequenceId() + 1).build());
     } else if (useDbMatcher.matches()) {
       sqlEngine.useDatabase(getConnectionId(ctx), useDbMatcher.group(1));
       ctx.writeAndFlush(OkResponse.builder().sequenceId(query.getSequenceId() + 1).build());

extensions/open-mysql-db/src/main/java/com/_4paradigm/openmldb/mysql/server/OpenmldbMysqlServer.java

@@ -10,6 +10,7 @@
 import com._4paradigm.openmldb.jdbc.SQLResultSet;
 import com._4paradigm.openmldb.mysql.mock.MockResult;
 import com._4paradigm.openmldb.mysql.util.TypeUtil;
+import com._4paradigm.openmldb.proto.NS;
 import com._4paradigm.openmldb.sdk.Column;
 import com._4paradigm.openmldb.sdk.Schema;
 import com._4paradigm.openmldb.sdk.SdkOption;
@@ -56,6 +57,12 @@ public class OpenmldbMysqlServer {
       Pattern.compile(
           "(?i)SELECT COUNT\\(\\*\\) FROM information_schema\\.TABLES WHERE TABLE_SCHEMA = '(.+)'");
 
+  // SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'xzs' AND table_name =
+  // 't_exam_paper'
+  private final Pattern selectCountSchemaTablesPattern =
+      Pattern.compile(
+          "(?i)SELECT COUNT\\(\\*\\) FROM information_schema\\.TABLES WHERE TABLE_SCHEMA = '(.+)' AND table_name = '(.+)'");
+
   // SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = 'xzs'
   private final Pattern selectCountColumnsPattern =
       Pattern.compile(
@@ -182,6 +189,10 @@ public void query(
                 return;
               }
 
+              if (mockSelectSchemaTableCount(connectionId, resultSetWriter, sql)) {
+                return;
+              }
+
               // This mock must execute before mockPatternQuery
               // SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'demo_db'
               // UNION SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA =
@@ -240,6 +251,7 @@ public void query(
                 return;
               }
 
+              String originalSql = sql;
               if (sql.startsWith("SHOW FULL TABLES")) {
                 // SHOW FULL TABLES WHERE Table_type != 'VIEW'
                 Matcher showTablesFromDbMatcher = showTablesFromDbPattern.matcher(sql);
@@ -250,6 +262,14 @@ public void query(
                 } else {
                   sql = "SHOW TABLES";
                 }
+              } else if (sql.matches("(?i)(?s)^\\s*CREATE TABLE.*$")) {
+                // convert data type TEXT to STRING
+                sql = sql.replaceAll("(?i) TEXT", " STRING");
+                // sql = sql.replaceAll("(?i) DATETIME", " DATE");
+                if (!sql.toLowerCase().contains(" not null")
+                    && sql.toLowerCase().contains(" null")) {
+                  sql = sql.replaceAll("(?i) null", "");
+                }
               } else {
                 Matcher crateDatabaseMatcher = createDatabasePattern.matcher(sql);
                 Matcher selectLimitMatcher = selectLimitPattern.matcher(sql);
@@ -264,7 +284,7 @@ public void query(
 
               if (sql.toLowerCase().startsWith("select") || sql.toLowerCase().startsWith("show")) {
                 SQLResultSet resultSet = (SQLResultSet) stmt.getResultSet();
-                outputResultSet(resultSetWriter, resultSet, sql);
+                outputResultSet(resultSetWriter, resultSet, originalSql);
               }
 
               System.out.println("Success to execute OpenMLDB SQL: " + sql);
@@ -622,6 +642,36 @@ private boolean mockPatternQuery(ResultSetWriter resultSetWriter, String sql) {
             return false;
           }
 
+          private boolean mockSelectSchemaTableCount(
+              int connectionId, ResultSetWriter resultSetWriter, String sql) throws SQLException {
+            // SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'xzs' AND
+            // table_name = 't_exam_paper'
+            Matcher selectCountSchemaTablesMatcher = selectCountSchemaTablesPattern.matcher(sql);
+            if (selectCountSchemaTablesMatcher.matches()) {
+              // COUNT(*)
+              List<QueryResultColumn> columns = new ArrayList<>();
+              columns.add(new QueryResultColumn("COUNT(*)", "VARCHAR(255)"));
+              resultSetWriter.writeColumns(columns);
+
+              List<String> row;
+              String dbName = selectCountSchemaTablesMatcher.group(1);
+              String tableName = selectCountSchemaTablesMatcher.group(2);
+              row = new ArrayList<>();
+              NS.TableInfo tableInfo =
+                  sqlClusterExecutorMap.get(connectionId).getTableInfo(dbName, tableName);
+              if (tableInfo == null || tableInfo.getName().equals("")) {
+                row.add("0");
+              } else {
+                row.add("1");
+              }
+              resultSetWriter.writeRow(row);
+
+              resultSetWriter.finish();
+              return true;
+            }
+            return false;
+          }
+
           private boolean mockSelectCountUnion(
               int connectionId, ResultSetWriter resultSetWriter, String sql) throws SQLException {
             // SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'demo_db'
@@ -713,14 +763,18 @@ public void outputResultSet(ResultSetWriter resultSetWriter, SQLResultSet result
     // Add schema
     for (int i = 0; i < columnCount; i++) {
       String columnName = schema.getColumnName(i);
-      if (sql.equalsIgnoreCase("show table status") && columnName.equalsIgnoreCase("table_id")) {
+      if ((sql.startsWith("SHOW FULL TABLES") || sql.equalsIgnoreCase("show table status"))
+          && columnName.equalsIgnoreCase("table_id")) {
         tableIdColumnIndex = i;
         continue;
       }
       int columnType = schema.getColumnType(i);
       columns.add(
           new QueryResultColumn(columnName, TypeUtil.openmldbTypeToMysqlTypeString(columnType)));
     }
+    if (sql.startsWith("SHOW FULL TABLES")) {
+      columns.add(new QueryResultColumn("Table_type", "VARCHAR(255)"));
+    }
 
     resultSetWriter.writeColumns(columns);
 
@@ -739,6 +793,9 @@ public void outputResultSet(ResultSetWriter resultSetWriter, SQLResultSet result
         String columnValue = TypeUtil.getResultSetStringColumn(resultSet, i + 1, type);
         row.add(columnValue);
       }
+      if (sql.startsWith("SHOW FULL TABLES")) {
+        row.add("BASE TABLE");
+      }
 
       resultSetWriter.writeRow(row);
     }