修正分页查询错误

sqlrest-common/src/main/java/com/gitee/sqlrest/common/dto/ProductContext.java

@@ -28,4 +28,5 @@ public class ProductContext {
   private boolean hasCatalogAndSchema;
   private Function<String, Pair<String, String>> adapter;
   private String pageSql;
+  private ThreeConsumer<Integer, Integer, List<Object>> pageConsumer;
 }

sqlrest-common/src/main/java/com/gitee/sqlrest/common/dto/ThreeConsumer.java

+package com.gitee.sqlrest.common.dto;
+
+@FunctionalInterface
+public interface ThreeConsumer<X, Y, Z> {
+
+  void accept(X arg1, Y arg2, Z arg3);
+}

sqlrest-common/src/main/java/com/gitee/sqlrest/common/enums/ProductTypeEnum.java

 package com.gitee.sqlrest.common.enums;
 
 import com.gitee.sqlrest.common.dto.ProductContext;
+import com.gitee.sqlrest.common.dto.ThreeConsumer;
 import java.util.Arrays;
 import java.util.Collections;
 import lombok.Getter;
@@ -28,7 +29,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT `SCHEMA_NAME` FROM `information_schema`.`SCHEMATA`")
           .adapter(database -> Pair.of(database, null))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
   /**
    * MariaDB数据库类型
    */
@@ -47,7 +53,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT `SCHEMA_NAME` FROM `information_schema`.`SCHEMATA`")
           .adapter(database -> Pair.of(database, null))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
   /**
    * Oracle数据库类型
    */
@@ -65,10 +76,16 @@ public enum ProductTypeEnum {
           .urlSample("jdbc:oracle:thin:@172.17.2.10:1521:ORCL")
           .sqlSchemaList("SELECT USERNAME FROM SYS.ALL_USERS")
           .adapter(database -> Pair.of(null, database))
-          .pageSql("SELECT * FROM ( SELECT ALIAS.*, ROWNUM ROW_ID FROM ( %s ) ALIAS WHERE ROWNUM <= ? ) WHERE ROW_ID > ?")
-          .build()),
+          .pageSql(
+              "SELECT * FROM ( SELECT ALIAS.*, ROWNUM ROW_ID FROM ( %s ) ALIAS WHERE ROWNUM <= ? ) WHERE ROW_ID > ?")
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(page * size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
   /**
-   * Microsoft SQL Server数据库类型(>=2005)
+   * Microsoft SQL Server数据库类型(>=2012)
    */
   SQLSERVER(
       ProductContext.builder()
@@ -85,7 +102,12 @@ public enum ProductTypeEnum {
           .hasCatalogAndSchema(true)
           .adapter(database -> Pair.of(null, database))
           .pageSql("SELECT * FROM (%s) ALIAS OFFSET ? ROWS FETCH NEXT ? ROWS ONLY")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add((page - 1) * size);
+                parameters.add(size);
+              }
+          ).build()),
 
   /**
    * PostgreSQL数据库类型
@@ -104,7 +126,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * DB2数据库类型
@@ -125,7 +152,12 @@ public enum ProductTypeEnum {
           .adapter(database -> Pair.of(null, database))
           .pageSql(
               "SELECT * FROM (SELECT TMP_PAGE.*,ROWNUMBER() OVER() AS ROW_ID FROM ( %s ) AS TMP_PAGE) ALIAS WHERE ROW_ID BETWEEN ? AND ?")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add((page - 1) * size);
+                parameters.add(page * size);
+              }
+          ).build()),
 
   /**
    * [国产] 达梦(DM)数据库类型
@@ -144,7 +176,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT DISTINCT object_name FROM ALL_OBJECTS WHERE OBJECT_TYPE = 'SCH'")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * [国产] 金仓(Kingbase)数据库类型
@@ -163,7 +200,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * [国产] 神通(Oscar)数据库类型
@@ -182,7 +224,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * [国产] 南大通用(GBase8A)数据库类型
@@ -201,7 +248,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * Sybase 数据库类型
@@ -220,7 +272,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("SELECT * FROM (%s) ALIAS OFFSET ? ROWS FETCH NEXT ? ROWS ONLY")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add((page - 1) * size);
+                parameters.add(size);
+              }
+          ).build()),
 
   /**
    * Hive 数据库类型
@@ -239,7 +296,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SHOW DATABASES")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * Sqlite v3数据库类型
@@ -260,7 +322,12 @@ public enum ProductTypeEnum {
           .retSchemaList(Collections.singletonList("sqlite_master"))
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * OpenGauss数据库类型
@@ -279,7 +346,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? offset ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * ClickHouse数据库类型
@@ -298,7 +370,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT schema_name FROM information_schema.schemata ")
           .adapter(database -> Pair.of(null, database))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * Doris数据库类型
@@ -318,7 +395,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT `SCHEMA_NAME` FROM `information_schema`.`SCHEMATA`")
           .adapter(database -> Pair.of(database, null))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
    * StarRocks数据库类型
@@ -338,10 +420,15 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT `SCHEMA_NAME` FROM `information_schema`.`SCHEMATA`")
           .adapter(database -> Pair.of(database, null))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
 
   /**
-   * OceanBase数据库类型
+   * OceanBase数据库类型(MySQL方言)
    */
   OCEANBASE(
       ProductContext.builder()
@@ -358,7 +445,12 @@ public enum ProductTypeEnum {
           .sqlSchemaList("SELECT `SCHEMA_NAME` FROM `information_schema`.`SCHEMATA`")
           .adapter(database -> Pair.of(database, null))
           .pageSql("select * from (%s) alias limit ? OFFSET ? ")
-          .build()),
+          .pageConsumer(
+              (page, size, parameters) -> {
+                parameters.add(size);
+                parameters.add((page - 1) * size);
+              }
+          ).build()),
   ;
 
   private ProductContext context;
@@ -399,6 +491,10 @@ public enum ProductTypeEnum {
     return this.context.getTestSql();
   }
 
+  public ThreeConsumer getPageConsumer() {
+    return this.context.getPageConsumer();
+  }
+
   public boolean hasDatabaseName() {
     return !Arrays.asList(DM, SQLITE3, MYSQL, MARIADB, GBASE8A).contains(this);
   }

sqlrest-core/src/main/java/com/gitee/sqlrest/core/exec/engine/impl/SqlExecutorService.java

 package com.gitee.sqlrest.core.exec.engine.impl;
 
-import cn.hutool.core.util.NumberUtil;
-import com.gitee.sqlrest.common.consts.Constants;
 import com.gitee.sqlrest.common.enums.NamingStrategyEnum;
 import com.gitee.sqlrest.common.enums.ProductTypeEnum;
 import com.gitee.sqlrest.core.exec.engine.AbstractExecutorEngine;
+import com.gitee.sqlrest.core.util.PageSizeUtils;
 import com.gitee.sqlrest.core.util.SqlJdbcUtils;
 import com.gitee.sqlrest.persistence.entity.ApiContextEntity;
 import com.gitee.sqlrest.template.Configuration;
@@ -35,12 +34,8 @@ public class SqlExecutorService extends AbstractExecutorEngine {
         for (ApiContextEntity sql : scripts) {
           SqlTemplate template = cfg.getTemplate(sql.getSqlText());
           SqlMeta sqlMeta = template.process(params);
-          int page = (null == params.get(Constants.PARAM_PAGE_NUMBER))
-              ? 1
-              : NumberUtil.parseInt(params.get(Constants.PARAM_PAGE_NUMBER).toString());
-          int size = (null == params.get(Constants.PARAM_PAGE_SIZE))
-              ? 10
-              : NumberUtil.parseInt(params.get(Constants.PARAM_PAGE_SIZE).toString());
+          int page = PageSizeUtils.getPageFromParams(params);
+          int size = PageSizeUtils.getSizeFromParams(params);
           dataList.add(SqlJdbcUtils.execute(productType, connection, sqlMeta, strategy, page, size));
         }
         connection.commit();

sqlrest-core/src/main/java/com/gitee/sqlrest/core/exec/module/DbVarModule.java

@@ -7,6 +7,7 @@ import com.gitee.sqlrest.common.enums.ProductTypeEnum;
 import com.gitee.sqlrest.core.exec.annotation.Comment;
 import com.gitee.sqlrest.core.exec.annotation.Module;
 import com.gitee.sqlrest.core.util.ConvertUtils;
+import com.gitee.sqlrest.core.util.PageSizeUtils;
 import com.gitee.sqlrest.template.Configuration;
 import com.gitee.sqlrest.template.SqlMeta;
 import com.gitee.sqlrest.template.SqlTemplate;
@@ -109,14 +110,9 @@ public class DbVarModule {
     SqlMeta sqlMeta = template.process(params);
     String pageSql = productType.getPageSql(sqlMeta.getSql());
     List<Object> parameters = sqlMeta.getParameter();
-    int page = (null == params.get(Constants.PARAM_PAGE_NUMBER))
-        ? 1
-        : NumberUtil.parseInt(params.get(Constants.PARAM_PAGE_NUMBER).toString());
-    int size = (null == params.get(Constants.PARAM_PAGE_SIZE))
-        ? 10
-        : NumberUtil.parseInt(params.get(Constants.PARAM_PAGE_SIZE).toString());
-    parameters.add(((page - 1) * size) < 0 ? 0 : (page - 1) * size);
-    parameters.add(size);
+    int page = PageSizeUtils.getPageFromParams(params);
+    int size = PageSizeUtils.getSizeFromParams(params);
+    this.productType.getPageConsumer().accept(page, size, parameters);
     return build(jdbcTemplate.queryForList(pageSql, parameters.toArray()));
   }
 

sqlrest-core/src/main/java/com/gitee/sqlrest/core/gateway/FirewallFilterService.java

 package com.gitee.sqlrest.core.gateway;
 
-import static com.gitee.sqlrest.common.consts.Constants.GATEWAY_APPLICATION_NAME;
-
-import com.gitee.sqlrest.common.dto.ResultEntity;
 import com.gitee.sqlrest.common.enums.OnOffEnum;
 import com.gitee.sqlrest.common.enums.WhiteBlackEnum;
 import com.gitee.sqlrest.core.dto.UpdateFirewallRulesRequest;
 import com.gitee.sqlrest.persistence.dao.FirewallRulesDao;
 import com.gitee.sqlrest.persistence.entity.FirewallRulesEntity;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -18,10 +14,7 @@ import javax.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.util.Strings;
-import org.springframework.cloud.client.ServiceInstance;
-import org.springframework.cloud.client.discovery.DiscoveryClient;
 import org.springframework.stereotype.Service;
-import org.springframework.web.client.RestTemplate;
 
 @Slf4j
 @Service
@@ -31,8 +24,6 @@ public class FirewallFilterService {
 
   @Resource
   private FirewallRulesDao firewallRulesDao;
-  @Resource
-  private DiscoveryClient discoveryClient;
 
   @PostConstruct
   public void refresh() {
@@ -68,18 +59,7 @@ public class FirewallFilterService {
 
   public void updateFirewallRules(UpdateFirewallRulesRequest request) {
     firewallRulesDao.update(request.getStatus(), request.getMode(), request.getAddresses());
-    refresh();
-
-    // 通过HTTP接口同步到所有的gateway节点(如何保证强一致性呢?)
-    RestTemplate restTemplate = new RestTemplate();
-    List<ServiceInstance> instances = discoveryClient.getInstances(GATEWAY_APPLICATION_NAME);
-    for (ServiceInstance instance : instances) {
-      String url = String.format("%s://%s:%s/cache/refresh",
-          instance.isSecure() ? "https" : "http",
-          instance.getHost(), instance.getPort());
-      restTemplate.getForEntity(url, ResultEntity.class);
-      log.info("sync ip rule cache to gateway node: {}", instance.getHost());
-    }
+    this.refresh();
   }
 
 }

sqlrest-core/src/main/java/com/gitee/sqlrest/core/util/PageSizeUtils.java

+package com.gitee.sqlrest.core.util;
+
+import cn.hutool.core.util.NumberUtil;
+import com.gitee.sqlrest.common.consts.Constants;
+import java.util.Map;
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class PageSizeUtils {
+
+  public static int getPageFromParams(Map<String, Object> params) {
+    int page = (null == params.get(Constants.PARAM_PAGE_NUMBER))
+        ? 1
+        : NumberUtil.parseInt(params.get(Constants.PARAM_PAGE_NUMBER).toString());
+    if (page <= 0) {
+      page = 1;
+    }
+    return page;
+  }
+
+  public static int getSizeFromParams(Map<String, Object> params) {
+    int size = (null == params.get(Constants.PARAM_PAGE_SIZE))
+        ? 10
+        : NumberUtil.parseInt(params.get(Constants.PARAM_PAGE_SIZE).toString());
+    if (size <= 0) {
+      size = 10;
+    }
+    return size;
+  }
+  
+}

sqlrest-core/src/main/java/com/gitee/sqlrest/core/util/SqlJdbcUtils.java

@@ -19,6 +19,8 @@ import lombok.extern.slf4j.Slf4j;
 @UtilityClass
 public class SqlJdbcUtils {
 
+  private static final int QUERY_TIMEOUT = 300;
+
   public static boolean isQuerySQL(String sql) {
     String upperSql = sql.toUpperCase().trim();
     return upperSql.startsWith("SELECT") || upperSql.startsWith("WITH");
@@ -29,23 +31,15 @@ public class SqlJdbcUtils {
   }
 
   public static Object execute(ProductTypeEnum productType, Connection connection, SqlMeta sqlMeta,
-      NamingStrategyEnum strategy, int page, int size)
-      throws SQLException {
+      NamingStrategyEnum strategy, int page, int size) throws SQLException {
     List<Object> paramValues = sqlMeta.getParameter();
     boolean isQuerySql = isQuerySQL(sqlMeta.getSql());
     String sql = isQuerySql ? productType.getPageSql(sqlMeta.getSql()) : sqlMeta.getSql();
     PreparedStatement statement = connection.prepareStatement(sql);
-    statement.setQueryTimeout(300);
-    statement.setFetchSize(isMySqlConnection(connection) ? Integer.MIN_VALUE : 100);
+    statement.setQueryTimeout(QUERY_TIMEOUT);
+    statement.setFetchSize(isMySqlConnection(connection) ? Integer.MIN_VALUE : size);
     if (isQuerySql) {
-      if (page <= 0) {
-        page = 1;
-      }
-      if (size <= 0) {
-        size = 10;
-      }
-      paramValues.add(size);
-      paramValues.add((page - 1) * size);
+      productType.getPageConsumer().accept(page, size, paramValues);
     }
     for (int i = 1; i <= paramValues.size(); i++) {
       statement.setObject(i, paramValues.get(i - 1));

sqlrest-gateway/src/main/java/com/gitee/sqlrest/gateway/controller/RefreshCacheController.java

-package com.gitee.sqlrest.gateway.controller;
-
-import com.gitee.sqlrest.common.dto.ResultEntity;
-import com.gitee.sqlrest.core.gateway.FirewallFilterService;
-import javax.annotation.Resource;
-import org.springframework.http.MediaType;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-@RestController
-@RequestMapping("/cache")
-public class RefreshCacheController {
-
-  @Resource
-  private FirewallFilterService firewallFilterService;
-
-  @PostMapping(value = "/refresh", produces = MediaType.APPLICATION_JSON_VALUE)
-  public ResultEntity refresh() {
-    firewallFilterService.refresh();
-    return ResultEntity.success();
-  }
-
-}

sqlrest-gateway/src/main/java/com/gitee/sqlrest/gateway/filter/FirewallRefreshService.java

+package com.gitee.sqlrest.gateway.filter;
+
+import com.gitee.sqlrest.core.gateway.FirewallFilterService;
+import javax.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class FirewallRefreshService {
+
+  @Resource
+  private FirewallFilterService firewallFilterService;
+
+  /*每30秒执行一次*/
+  @EventListener(ApplicationReadyEvent.class)
+  @Scheduled(cron = "${cron.firewall.expression:0/30 0 * * * ?}")
+  public void loadFlowRules() {
+    try {
+      firewallFilterService.refresh();
+    } catch (Exception e) {
+      log.error("load firewall rules failed:{}", e.getMessage(), e);
+    }
+  }
+}