content: /**
   * DBSyncer Copyright 2020-2023 All Rights Reserved.
   */
  package org.dbsyncer.connector.postgresql;
  
  import org.dbsyncer.common.util.StringUtil;
  import org.dbsyncer.connector.postgresql.cdc.PostgreSQLListener;
  import org.dbsyncer.connector.postgresql.schema.PostgreSQLBitValueMapper;
  import org.dbsyncer.connector.postgresql.schema.PostgreSQLOtherValueMapper;
  import org.dbsyncer.connector.postgresql.schema.PostgreSQLSchemaResolver;
  import org.dbsyncer.connector.postgresql.validator.PostgreSQLConfigValidator;
  import org.dbsyncer.sdk.connector.ConfigValidator;
  import org.dbsyncer.sdk.connector.database.AbstractDatabaseConnector;
  import org.dbsyncer.sdk.constant.DatabaseConstant;
  import org.dbsyncer.sdk.enums.ListenerTypeEnum;
  import org.dbsyncer.sdk.listener.DatabaseQuartzListener;
  import org.dbsyncer.sdk.listener.Listener;
  import org.dbsyncer.sdk.model.PageSql;
  import org.dbsyncer.sdk.plugin.ReaderContext;
  import org.dbsyncer.sdk.schema.SchemaResolver;
  import org.dbsyncer.sdk.util.PrimaryKeyUtil;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.sql.Types;
  import java.util.List;
  
  /**
   * PostgreSQL连接器实现
   *
   * @Author AE86
   * @Version 1.0.0
   * @Date 2022-05-22 22:56
   */
  public final class PostgreSQLConnector extends AbstractDatabaseConnector {
  
      private final Logger logger = LoggerFactory.getLogger(getClass());
  
      private final PostgreSQLConfigValidator configValidator = new PostgreSQLConfigValidator();
  
      private final PostgreSQLSchemaResolver schemaResolver = new PostgreSQLSchemaResolver();
  
      public PostgreSQLConnector() {
          VALUE_MAPPERS.put(Types.BIT, new PostgreSQLBitValueMapper());
          VALUE_MAPPERS.put(Types.OTHER, new PostgreSQLOtherValueMapper());
      }
  
      @Override
      public String getConnectorType() {
          return "PostgreSQL";
      }
  
      @Override
      public ConfigValidator getConfigValidator() {
          return configValidator;
      }
  
      @Override
      public Listener getListener(String listenerType) {
          if (ListenerTypeEnum.isTiming(listenerType)) {
              return new DatabaseQuartzListener();
          }
  
          if (ListenerTypeEnum.isLog(listenerType)) {
              return new PostgreSQLListener();
          }
          return null;
      }
  
      @Override
      public String buildSqlWithQuotation() {
          return "\"";
      }
  
      @Override
      public String getPageSql(PageSql config) {
          // select * from test."my_user" where "id" > ? and "uid" > ? order by "id","uid" limit ? OFFSET ?
          StringBuilder sql = new StringBuilder(config.getQuerySql());
          if (PrimaryKeyUtil.isSupportedCursor(config.getFields())) {
              appendOrderByPk(config, sql);
          }
          sql.append(DatabaseConstant.POSTGRESQL_PAGE_SQL);
          return sql.toString();
      }
  
      @Override
      public String getPageCursorSql(PageSql config) {
          // 不支持游标查询
          if (!PrimaryKeyUtil.isSupportedCursor(config.getFields())) {
              logger.debug("不支持游标查询，主键包含非数字类型");
              return StringUtil.EMPTY;
          }
  
          // select * from test."my_user" where "id" > ? and "uid" > ? order by "id","uid" limit ? OFFSET ?
          StringBuilder sql = new StringBuilder(config.getQuerySql());
          boolean skipFirst = false;
          // 没有过滤条件
          if (StringUtil.isBlank(config.getQueryFilter())) {
              skipFirst = true;
              sql.append(" WHERE ");
          }
          final List<String> primaryKeys = config.getPrimaryKeys();
          final String quotation = buildSqlWithQuotation();
          PrimaryKeyUtil.buildSql(sql, primaryKeys, quotation, " AND ", " > ? ", skipFirst);
          appendOrderByPk(config, sql);
          sql.append(DatabaseConstant.POSTGRESQL_PAGE_SQL);
          return sql.toString();
      }
  
      @Override
      public Object[] getPageArgs(ReaderContext context) {
          int pageIndex = context.getPageIndex();
          int pageSize = context.getPageSize();
          return new Object[]{pageSize, (pageIndex - 1) * pageSize};
      }
  
      @Override
      public Object[] getPageCursorArgs(ReaderContext context) {
          int pageSize = context.getPageSize();
          Object[] cursors = context.getCursors();
          if (null == cursors) {
              return new Object[]{pageSize, 0};
          }
          int cursorsLen = cursors.length;
          Object[] newCursors = new Object[cursorsLen + 2];
          System.arraycopy(cursors, 0, newCursors, 0, cursorsLen);
          newCursors[cursorsLen] = pageSize;
          newCursors[cursorsLen + 1] = 0;
          return newCursors;
      }
  
      @Override
      public boolean enableCursor() {
          return true;
      }
  
      @Override
      public SchemaResolver getSchemaResolver() {
          return schemaResolver;
      }
  }