src/org/redkale/service/DataSQLListenerService.java

@@ -122,7 +122,7 @@ public class DataSQLListenerService implements DataSQLListener, Service {
 
     @MultiRun
     public void send(String... sqls) {
-        ((DataDefaultSource) source).execute(sqls);
+        ((DataDefaultSource) source).directExecute(sqls);
     }
 
 }

src/org/redkale/service/DataSourceService.java

@@ -7,7 +7,9 @@ package org.redkale.service;
 
 import java.io.*;
 import java.nio.channels.*;
+import java.sql.*;
 import java.util.*;
+import java.util.function.*;
 import javax.annotation.*;
 import org.redkale.source.*;
 import org.redkale.util.*;
@@ -545,4 +547,15 @@ public class DataSourceService implements DataSource, Service, AutoCloseable {
     public void close() throws Exception {
         source.getClass().getMethod("close").invoke(source);
     }
+
+    @Override
+    public final void directQuery(String sql, Consumer<ResultSet> consumer) {
+        source.directQuery(sql, consumer);
+    }
+
+    @Override
+    public final int[] directExecute(String... sqls) {
+        return source.directExecute(sqls);
+    }
+
 }

src/org/redkale/source/DataDefaultSource.java

@@ -255,23 +255,28 @@ public final class DataDefaultSource implements DataSource, Function<Class, Enti
         }
     }
 
-    public <T> void execute(String... sqls) {
+    @Override
+    public final int[] directExecute(String... sqls) {
         Connection conn = createWriteSQLConnection();
         try {
-            execute(conn, sqls);
+            return directExecute(conn, sqls);
         } finally {
             closeSQLConnection(conn);
         }
     }
 
-    private <T> void execute(final Connection conn, String... sqls) {
-        if (sqls.length == 0) return;
+    private int[] directExecute(final Connection conn, String... sqls) {
+        if (sqls.length == 0) return new int[0];
         try {
             final Statement stmt = conn.createStatement();
+            final int[] rs = new int[sqls.length];
+            int i = -1;
             for (String sql : sqls) {
-                stmt.execute(sql);
+                rs[++i] = stmt.execute(sql) ? 1 : 0;
+
             }
             stmt.close();
+            return rs;
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }
@@ -1712,4 +1717,22 @@ public final class DataDefaultSource implements DataSource, Function<Class, Enti
             closeSQLConnection(conn);
         }
     }
+
+    @Override
+    public void directQuery(String sql, Consumer<ResultSet> consumer) {
+        final Connection conn = createReadSQLConnection();
+        try {
+            if (debug.get()) logger.finest("queryDirect sql=" + sql);
+            final PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
+            final ResultSet set = ps.executeQuery();
+            consumer.accept(set);
+            set.close();
+            ps.close();
+        } catch (Exception ex) {
+            throw new RuntimeException(ex);
+        } finally {
+            closeSQLConnection(conn);
+        }
+    }
+
 }

src/org/redkale/source/DataSource.java

@@ -7,7 +7,9 @@ package org.redkale.source;
 
 import java.io.*;
 import java.nio.channels.*;
+import java.sql.*;
 import java.util.*;
+import java.util.function.*;
 import org.redkale.util.*;
 
 /**
@@ -302,4 +304,22 @@ public interface DataSource {
 
     public <T> void querySheet(final CompletionHandler<Sheet<T>, FilterNode> handler, final Class<T> clazz, final SelectColumn selects, final Flipper flipper, final FilterNode node);
 
+    //-----------------------direct----------------------------
+    /**
+     * 直接本地执行SQL语句进行查询，远程模式不可用
+     * 通常用于复杂的关联查询
+     *
+     * @param sql      SQL语句
+     * @param consumer 回调函数
+     */
+    public void directQuery(String sql, final Consumer<ResultSet> consumer);
+
+    /**
+     * 直接本地执行SQL语句进行增删改操作，远程模式不可用
+     * 通常用于复杂的更新操作
+     *
+     * @param sqls SQL语句
+     * @return 结果数组
+     */
+    public int[] directExecute(String... sqls);
 }