Java：MultiDataSource 为 APIJSON 和 SQLAuto 新增支持 Cassandra, Snowflake, Databricks

TommyLemon · TommyLemon · commit 4990d232b2bd · 2023-03-30T23:28:12.000+08:00
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/pom.xml b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/pom.xml
@@ -116,6 +116,21 @@
             <artifactId>influxdb-java</artifactId>
             <version>2.23</version>
         </dependency>
+        <dependency>
+            <groupId>net.snowflake</groupId>
+            <artifactId>snowflake-jdbc</artifactId>
+            <version>3.0.1</version>
+        </dependency>
+        <dependency>
+            <groupId>com.datastax.oss</groupId>
+            <artifactId>java-driver-core</artifactId>
+            <version>4.15.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.databricks</groupId>
+            <artifactId>databricks-jdbc</artifactId>
+            <version>2.6.25-1</version>
+        </dependency>
         <!-- Oracle, SQLServer 等其它数据库的 JDBC 驱动，可以在这里加上 Maven 依赖或 libs 目录放 Jar 包并依赖 -->
         <!-- 数据库 JDBC 驱动 >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> -->
 
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/boot/DemoApplication.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/boot/DemoApplication.java
@@ -100,7 +100,7 @@ public static void main(String[] args) throws Exception {
     // SpringBoot 2.x 自定义端口方式
     @Override
     public void customize(ConfigurableServletWebServerFactory server) {
-        server.setPort(8080);
+        server.setPort(9090);
     }
 
     // 支持 APIAuto 中 JavaScript 代码跨域请求
@@ -171,6 +171,26 @@ public void addCorsMappings(CorsRegistry registry) {
             Log.e(TAG, "加载 Dameng 驱动失败，请检查 pom.xml 中 com.dameng 版本是否存在以及可用 ！！！");
         }
 
+        try { //加载驱动程序
+            Log.d(TAG, "尝试加载 Snowflake 驱动 <<<<<<<<<<<<<<<<<<<<< ");
+            Class.forName("net.snowflake.client.jdbc.SnowflakeDriver");
+            Log.d(TAG, "成功加载 Snowflake 驱动！>>>>>>>>>>>>>>>>>>>>> ");
+        }
+        catch (ClassNotFoundException e) {
+            e.printStackTrace();
+            Log.e(TAG, "加载 Snowflake 驱动失败，请检查 pom.xml 中 net.snowflake 版本是否存在以及可用 ！！！");
+        }
+
+        try { //加载驱动程序
+            Log.d(TAG, "尝试加载 Databricks 驱动 <<<<<<<<<<<<<<<<<<<<< ");
+            Class.forName("com.databricks.client.jdbc.Driver");
+            Log.d(TAG, "成功加载 Databricks 驱动！>>>>>>>>>>>>>>>>>>>>> ");
+        }
+        catch (ClassNotFoundException e) {
+            e.printStackTrace();
+            Log.e(TAG, "加载 Databricks 驱动失败，请检查 pom.xml 中 com.databricks 版本是否存在以及可用 ！！！");
+        }
+
         //    try { //加载驱动程序
         //      Log.d(TAG, "尝试加载 TDengine 驱动 <<<<<<<<<<<<<<<<<<<<< ");
         //      Class.forName("com.taosdata.jdbc.TSDBDriver");
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/boot/DemoController.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/boot/DemoController.java
@@ -1373,12 +1373,16 @@ public String execute(@RequestBody String request, HttpSession session) {
             JSONObject req = JSON.parseObject(request);
             String sql = req == null ? null : req.getString("sql");
             if (StringUtil.isEmpty(sql)) {
-                throw new IllegalArgumentException("SQL 不能为空！");
+                throw new IllegalArgumentException("sql 不能为空！");
             }
 
             String database = req.getString("database");
-            String schema = req.getString("schema");
             String uri = req.getString("uri");
+            if (StringUtil.isEmpty(database, true) && StringUtil.isEmpty(uri, true)) {
+                throw new NullPointerException("database 和 uri 不能同时为空！至少其中一个要传有效值!");
+            }
+            String schema = req.getString("schema");
+
             String account = req.getString("account");
             String password = req.getString("password");
             JSONArray arg = req.getJSONArray("args");
@@ -1392,7 +1396,7 @@ public String execute(@RequestBody String request, HttpSession session) {
                         prefix = prefix.substring(mid + 1);
                     }
 
-                    if (DemoSQLConfig.DATABASE_INFLUXDB.equalsIgnoreCase(prefix)) {
+                    if (DemoSQLConfig.DATABASE_INFLUXDB.equalsIgnoreCase(prefix) || DemoSQLExecutor.DATABASE_CASSANDRA.equalsIgnoreCase(prefix)) {
                         database = prefix.toUpperCase();
 
                         int end = uri.lastIndexOf("/");
@@ -1403,10 +1407,16 @@ public String execute(@RequestBody String request, HttpSession session) {
                             uri = uri.substring(0, end);
                         }
 
-                        uri = "http" + uri.substring(start);
+//                        if (DemoSQLConfig.DATABASE_INFLUXDB.equalsIgnoreCase(prefix)) {
+                            uri = "http" + uri.substring(start);
+                            // account = "root";
+                            // password = "apijson@123";
+//                        }
 
-//                        account = "root";
-//                        password = "apijson@123";
+                        if (DemoSQLExecutor.DATABASE_CASSANDRA.equalsIgnoreCase(prefix)) { // unknown protocol: jdbc
+                            account = "test"; // "peter";
+                            password = "test"; // null;
+                        }
                     } else if (DemoSQLExecutor.DATABASE_NEBULA.equalsIgnoreCase(prefix)) {
                         database = prefix.toUpperCase();
                     }
@@ -1440,7 +1450,7 @@ public String execute(@RequestBody String request, HttpSession session) {
                 long rsDuration = 0;
                 long executeStartTime = System.currentTimeMillis();
 
-                if (DemoSQLConfig.DATABASE_INFLUXDB.equals(database)) {
+                if (DemoSQLConfig.DATABASE_INFLUXDB.equals(database) || DemoSQLExecutor.DATABASE_CASSANDRA.equals(database)) {
                     JSONObject result = executor.execute(config, false);
                     executeDuration = System.currentTimeMillis() - executeStartTime;
 
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoFunctionParser.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoFunctionParser.java
@@ -22,6 +22,7 @@
 
 import javax.servlet.http.HttpSession;
 
+import apijson.orm.script.JavaScriptExecutor;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 
@@ -42,6 +43,10 @@
 public class DemoFunctionParser extends APIJSONFunctionParser {
 	public static final String TAG = "DemoFunctionParser";
 
+	static {
+		SCRIPT_EXECUTOR_MAP.put("js", new JavaScriptExecutor());
+	}
+
 	public DemoFunctionParser() {
 		this(null, null, 0, null, null);
 	}
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLConfig.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLConfig.java
@@ -260,7 +260,7 @@ public String getDBAccount() {
 			return "root";
 		}
 
-		return "";
+		return null;
 	}
 
 	private String dbPassword;
@@ -303,7 +303,7 @@ public String getDBPassword() {
 			return "apijson@123"; //TODO 改成你自己的
 		}
 
-		return "";
+		return null;
 	}
 
 	private String sql;
diff --git a/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLExecutor.java b/APIJSON-Java-Server/APIJSONBoot-MultiDataSource/src/main/java/apijson/demo/DemoSQLExecutor.java
@@ -17,10 +17,17 @@
 import apijson.*;
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.fastjson.JSONObject;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
 import com.vesoft.nebula.jdbc.impl.NebulaDriver;
 import com.zaxxer.hikari.HikariDataSource;
 
 import java.io.Serializable;
+import java.net.URI;
+import java.net.URL;
+import java.nio.file.Paths;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.Collection;
@@ -37,7 +44,6 @@
 import org.influxdb.BatchOptions;
 import org.influxdb.InfluxDB;
 import org.influxdb.InfluxDBFactory;
-import org.influxdb.dto.Point;
 import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
@@ -94,7 +100,7 @@ public synchronized void putCache(String sql, List<JSONObject> list, SQLConfig c
         super.putCache(sql, list, config);
 
         String table = config != null && config.isMain() ? config.getTable() : null;
-        if (table != null && DemoSQLConfig.CONFIG_TABLE_LIST.contains(table) == false) {
+        if (table != null && ! DemoSQLConfig.CONFIG_TABLE_LIST.contains(table)) {
             try {
                 if (config.isExplain() || RequestMethod.isHeadMethod(config.getMethod(), true)) {
                     REDIS_TEMPLATE.opsForValue().set(sql, JSON.toJSONString(list), 10 * 60, TimeUnit.SECONDS);
@@ -124,6 +130,8 @@ public synchronized void removeCache(String sql, SQLConfig config) {
     // Redis 缓存 >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
 
     public static final String DATABASE_NEBULA = "NEBULA";
+    public static final String DATABASE_SNOWFLAKE = "SNOWFLAKE";
+    public static final String DATABASE_CASSANDRA = "CASSANDRA";
 
     // 适配连接池，如果这里能拿到连接池的有效 Connection，则 SQLConfig 不需要配置 dbVersion, dbUri, dbAccount, dbPassword
     @Override
@@ -196,82 +204,123 @@ public Connection getConnection(SQLConfig config) throws Exception {
 
     @Override
     public JSONObject execute(@NotNull SQLConfig config, boolean unknownType) throws Exception {
-        if (DemoSQLConfig.DATABASE_INFLUXDB.equals(config.getDatabase())) {
-            InfluxDB influxDB = InfluxDBFactory.connect(config.getDBUri(), config.getDBAccount(), config.getDBPassword());
-            influxDB.setDatabase(config.getSchema());
+        if (DATABASE_CASSANDRA.equals(config.getDatabase()) || DemoSQLConfig.DATABASE_INFLUXDB.equals(config.getDatabase())) {
 
             String sql = config.getSQL(config.isPrepared());
 
             RequestMethod method = config.getMethod();
-            boolean isWrite = ! RequestMethod.isQueryMethod(method);
-            if (method == null && ! isWrite) {
+            boolean isWrite = !RequestMethod.isQueryMethod(method);
+            if (method == null && !isWrite) {
                 String trimmedSQL = sql == null ? null : sql.trim();
                 String sqlPrefix = trimmedSQL == null || trimmedSQL.length() < 7 ? "" : trimmedSQL.substring(0, 7).toUpperCase();
                 isWrite = sqlPrefix.startsWith("INSERT ") || sqlPrefix.startsWith("UPDATE ") || sqlPrefix.startsWith("DELETE ");
             }
 
-            if (isWrite) {
-                influxDB.enableBatch(
-                        BatchOptions.DEFAULTS
-                                .threadFactory(runnable -> {
-                                    Thread thread = new Thread(runnable);
-                                    thread.setDaemon(true);
-                                    return thread;
-                                })
-                );
 
-                Runtime.getRuntime().addShutdownHook(new Thread(influxDB::close));
+            if (DATABASE_CASSANDRA.equals(config.getDatabase())) {
+                CqlSession session = CqlSession.builder()
+//                        .withCloudSecureConnectBundle(Paths.get("/path/to/secure-connect-database_name.zip"))
+                        .withCloudSecureConnectBundle(new URL(config.getDBUri()))
+                        .withAuthCredentials(config.getDBAccount(), config.getDBPassword())
+                        .withKeyspace(config.getSchema())
+                        .build();
+
+                //            if (config.isPrepared()) {
+                //                PreparedStatement stt = session.prepare(sql);
+                //
+                //                List<Object> pl = config.getPreparedValueList();
+                //                if (pl != null) {
+                //                    for (Object o : pl) {
+                //                        stt.bind(pl.toArray());
+                //                    }
+                //                }
+                //                sql = stt.getQuery();
+                //            }
+
+                ResultSet rs = session.execute(sql);
+
+                List<Row> list = rs.all();
+                if (list == null || list.isEmpty()) {
+                    return new JSONObject(true);
+                }
 
-                influxDB.write(sql);
+                JSONObject result = JSON.parseObject(list.get(0));
+                if (list.size() > 1) {
+                    result.put(KEY_RAW_LIST, list);
+                }
 
-                JSONObject result = DemoParser.newSuccessResult();
+                return result;
+            }
 
-                if (method == RequestMethod.POST) {
-                    List<List<Object>> values = config.getValues();
-                    result.put(JSONResponse.KEY_COUNT, values == null ? 0 : values.size());
-                } else {
-                    String idKey = config.getIdKey();
-                    Object id = config.getId();
-                    Object idIn = config.getIdIn();
-                    if (id != null) {
-                        result.put(idKey, id);
-                    }
-                    if (idIn != null) {
-                        result.put(idKey + "[]", idIn);
-                    }
 
-                    if (method == RequestMethod.PUT) {
-                        Map<String, Object> content = config.getContent();
-                        result.put(JSONResponse.KEY_COUNT, content == null ? 0 : content.size());
+            if (DemoSQLConfig.DATABASE_INFLUXDB.equals(config.getDatabase())) {
+                InfluxDB influxDB = InfluxDBFactory.connect(config.getDBUri(), config.getDBAccount(), config.getDBPassword());
+                influxDB.setDatabase(config.getSchema());
+
+
+                if (isWrite) {
+                    influxDB.enableBatch(
+                            BatchOptions.DEFAULTS
+                                    .threadFactory(runnable -> {
+                                        Thread thread = new Thread(runnable);
+                                        thread.setDaemon(true);
+                                        return thread;
+                                    })
+                    );
+
+                    Runtime.getRuntime().addShutdownHook(new Thread(influxDB::close));
+
+                    influxDB.write(sql);
+
+                    JSONObject result = DemoParser.newSuccessResult();
+
+                    if (method == RequestMethod.POST) {
+                        List<List<Object>> values = config.getValues();
+                        result.put(JSONResponse.KEY_COUNT, values == null ? 0 : values.size());
                     } else {
-                        result.put(JSONResponse.KEY_COUNT, id == null && idIn instanceof Collection ? ((Collection<?>) idIn).size() : 1); // FIXME 直接 SQLAuto 传 Flux/InfluxQL INSERT 如何取数量？
+                        String idKey = config.getIdKey();
+                        Object id = config.getId();
+                        Object idIn = config.getIdIn();
+                        if (id != null) {
+                            result.put(idKey, id);
+                        }
+                        if (idIn != null) {
+                            result.put(idKey + "[]", idIn);
+                        }
+
+                        if (method == RequestMethod.PUT) {
+                            Map<String, Object> content = config.getContent();
+                            result.put(JSONResponse.KEY_COUNT, content == null ? 0 : content.size());
+                        } else {
+                            result.put(JSONResponse.KEY_COUNT, id == null && idIn instanceof Collection ? ((Collection<?>) idIn).size() : 1); // FIXME 直接 SQLAuto 传 Flux/InfluxQL INSERT 如何取数量？
+                        }
                     }
+
+                    return result;
                 }
 
-                return result;
-            }
+                QueryResult qr = influxDB.query(new Query(sql));
 
-            QueryResult qr = influxDB.query(new Query(sql));
+                String err = qr == null ? null : qr.getError();
+                if (StringUtil.isNotEmpty(qr, true)) {
+                    throw new SQLException(err);
+                }
 
-            String err = qr == null ? null : qr.getError();
-            if (StringUtil.isNotEmpty(qr, true)) {
-                throw new SQLException(err);
-            }
+                List<QueryResult.Result> list = qr == null ? null : qr.getResults();
+                if (list == null || list.isEmpty()) {
+                    return new JSONObject(true);
+                }
 
-            List<QueryResult.Result> list = qr == null ? null : qr.getResults();
-            if (list == null || list.isEmpty()) {
-                return new JSONObject(true);
-            }
+                JSONObject result = JSON.parseObject(list.get(0));
+                if (list.size() > 1) {
+                    result.put(KEY_RAW_LIST, list);
+                }
 
-            JSONObject result = JSON.parseObject(list.get(0));
-            if (list.size() > 1) {
-                result.put(KEY_RAW_LIST, list);
+                return result;
             }
 
-            return result;
         }
 
-
         return super.execute(config, unknownType);
     }
 

Original file line number	Diff line number	Diff line change
`@@ -260,7 +260,7 @@ public String getDBAccount() {`
`260`	`260`	`return "root";`
`261`	`261`	`}`
`262`	`262`
`263`		`- return "";`
	`263`	`+ return null;`
`264`	`264`	`}`
`265`	`265`
`266`	`266`	`private String dbPassword;`
`@@ -303,7 +303,7 @@ public String getDBPassword() {`
`303`	`303`	`return "apijson@123"; //TODO 改成你自己的`
`304`	`304`	`}`
`305`	`305`
`306`		`- return "";`
	`306`	`+ return null;`
`307`	`307`	`}`
`308`	`308`
`309`	`309`	`private String sql;`