content: package com.alibaba.otter.canal.client.adapter.phoenix.service;
  
  import java.sql.*;
  import java.util.*;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicLong;
  
  import javax.sql.DataSource;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.canal.client.adapter.phoenix.PhoenixAdapter;
  import com.alibaba.otter.canal.client.adapter.phoenix.config.MappingConfig;
  import com.alibaba.otter.canal.client.adapter.phoenix.config.MappingConfig.DbMapping;
  import com.alibaba.otter.canal.client.adapter.phoenix.support.PhoenixSupportUtil;
  import com.alibaba.otter.canal.client.adapter.phoenix.support.SyncUtil;
  import com.alibaba.otter.canal.client.adapter.phoenix.support.TypeUtil;
  import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
  import com.alibaba.otter.canal.client.adapter.support.EtlResult;
  import com.alibaba.otter.canal.client.adapter.support.Util;
  import com.google.common.base.Joiner;
  
  /**
   * Phoenix ETL 操作业务类
   */
  public class PhoenixEtlService {
  
      private static final Logger logger = LoggerFactory.getLogger(PhoenixEtlService.class);
  
      private static String[] splitNotEmpty(String s) {
          if (s != null && s.trim().length() > 0) {
              return s.trim().split(",");
          }
          return new String[]{};
      }
  
  
      static boolean syncSchema(Connection targetDSConnection, MappingConfig config) {
          DataSource srcDataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
          if (srcDataSource == null) {
              return false;
          }
  
          try (Connection conn = srcDataSource.getConnection()) {
              return syncSchema(conn, targetDSConnection, config);
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
      }
  
      private static boolean syncSchema(DataSource srcDS,Connection targetDSConnection, MappingConfig config) {
          try (Connection conn = srcDS.getConnection()) {
              return syncSchema(conn, targetDSConnection, config);
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
      }
  
      @SuppressWarnings("unchecked")
      private static boolean syncSchema(Connection srcDS, Connection targetDS, MappingConfig config) {
          DbMapping dbMapping = config.getDbMapping();
          if (dbMapping.getMapAll() && dbMapping.isAlter()) { // 检查字段是否缺失
              Map<String, Integer> targetColumnType = new LinkedHashMap<>();
              String targetTable = SyncUtil.getDbTableName(dbMapping);
              try {
                  Util.sqlRS(targetDS, "SELECT * FROM " + targetTable + " LIMIT 1", rs -> {
                      try {
                          ResultSetMetaData rsd = rs.getMetaData();
                          int columnCount = rsd.getColumnCount();
                          for (int i = 1; i <= columnCount; i++) {
                              targetColumnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
                          }
                      } catch (Exception e) {
                          logger.error(dbMapping.getTable() + " etl failed! ==>" + e.getMessage(), e);
                      }
                  });
              } catch (RuntimeException e) {
                  if (!e.getCause().getClass().getName().endsWith("TableNotFoundException")) {
                      throw e;
                  }
              }
              StringBuilder missing = new StringBuilder();
              StringBuilder constraint = new StringBuilder();
              Util.sqlRS(srcDS, "SELECT * FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = '" + dbMapping.getDatabase() + "'  AND TABLE_NAME = '" + dbMapping.getTable() + "'", rs -> {
                  try {
                      List<String> excludeColumns = config.getDbMapping().getExcludeColumns();
                      while (rs.next()) {
                          String name = rs.getString("COLUMN_NAME");
                          String lower = name.toLowerCase();
                          String colType = rs.getString("COLUMN_TYPE");
                          if (targetColumnType.get(lower) == null && !excludeColumns.contains(lower)) {
                              boolean isPri = rs.getString("COLUMN_KEY").equals("PRI");
                              String[] args = splitNotEmpty(colType.replaceAll("^\\w+(?:\\(([^)]*)\\))?[\\s\\S]*$", "$1"));
                              missing.append(dbMapping.escape(name)).append(" ").append(TypeUtil.getPhoenixType(
                                      rs.getString("DATA_TYPE").toUpperCase(),
                                      args,
                                      colType.contains("unsigned"),
                                      dbMapping.isLimit()
                              ));
                              if (isPri) {
                                  if (args.length > 0 && dbMapping.isLimit() || rs.getString("IS_NULLABLE").equals("NO")) {
                                      missing.append(" NOT NULL");
                                  }
                                  constraint.append(dbMapping.escape(name)).append(',');
                              }
                              missing.append(',');
                          }
                      }
                  } catch (Exception e) {
                      logger.error(dbMapping.getDatabase() + "." + dbMapping.getTable() + " schema failed! ==>" + e.getMessage(), e);
                      throw new RuntimeException(e);
                  }
              });
              if (missing.length() > 0) {
                  String sql;
                  if (targetColumnType.isEmpty()) {
                      if (constraint.length() > 0) {
                          constraint.deleteCharAt(constraint.length() - 1);
                          missing.append("CONSTRAINT pk PRIMARY KEY(").append(constraint.toString()).append(")");
                      } else {
                          missing.deleteCharAt(missing.length() - 1);
                      }
                      sql = "CREATE TABLE " + targetTable + " (" + missing.toString() + ")";
                  } else {
                      missing.deleteCharAt(missing.length() - 1);
                      sql = "ALTER TABLE " + targetTable + " ADD " + missing.toString();
                  }
                  logger.info("schema missing: {} {}", targetColumnType, sql);
                  try (PreparedStatement pstmt = targetDS.prepareStatement(sql)) {
                      pstmt.executeUpdate();
                  } catch (SQLException e) {
                      logger.error("sync schema error: " + e.getMessage(), e);
                  }
              } else {
                  logger.debug("schema ok: {}", targetColumnType);
              }
              return true;
          }
          return false;
      }
  
      /**
       * 导入数据
       */
      public static EtlResult importData(DataSource srcDS, Connection targetDSConnection, MappingConfig config,
                                         List<String> params) {
          EtlResult etlResult = new EtlResult();
          AtomicLong successCount = new AtomicLong();
          List<String> errMsg = new ArrayList<>();
          String hbaseTable = "";
          try {
              if (config == null) {
                  logger.error("Config is null!");
                  etlResult.setSucceeded(false);
                  etlResult.setErrorMessage("Config is null!");
                  return etlResult;
              }
              boolean debug = params != null && params.get(0).equals("_debug");
              if (debug) {
                  params = params.subList(1, params.size());
              }
              syncSchema(srcDS, targetDSConnection, config);
              DbMapping dbMapping = config.getDbMapping();
  
              long start = System.currentTimeMillis();
  
              // 拼接sql
              StringBuilder sql = new StringBuilder(
                      "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable());
  
              // 拼接条件
              appendCondition(params, dbMapping, srcDS, sql);
  
              // 获取总数
              String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
              long cnt = (Long) Util.sqlRS(srcDS, countSql, rs -> {
                  Long count = null;
                  try {
                      if (rs.next()) {
                          count = ((Number) rs.getObject(1)).longValue();
                      }
                  } catch (Exception e) {
                      logger.error(e.getMessage(), e);
                  }
                  return count == null ? 0 : count;
              });
  
              // 当大于1万条记录时开启多线程
              if (cnt >= 10000) {
                  int threadCount = 3;
                  long perThreadCnt = cnt / threadCount;
                  ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
                  for (int i = 0; i < threadCount; i++) {
                      long offset = i * perThreadCnt;
                      Long size = null;
                      if (i != threadCount - 1) {
                          size = perThreadCnt;
                      }
                      String sqlFinal;
                      if (size != null) {
                          sqlFinal = sql + " LIMIT " + offset + "," + size;
                      } else {
                          sqlFinal = sql + " LIMIT " + offset + "," + cnt;
                      }
                      executor
                              .execute(() -> executeSqlImport(srcDS, targetDSConnection, sqlFinal, dbMapping, successCount, errMsg, debug));
                  }
  
                  executor.shutdown();
                  //noinspection StatementWithEmptyBody
                  while (!executor.awaitTermination(3, TimeUnit.SECONDS)) ;
              } else {
                  executeSqlImport(srcDS, targetDSConnection, sql.toString(), dbMapping, successCount, errMsg, debug);
              }
  
              logger.info(
                      dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!");
  
              etlResult
                      .setResultMessage("导入目标表 " + SyncUtil.getDbTableName(dbMapping) + " 数据：" + successCount.get() + " 条");
          } catch (Exception e) {
              logger.error(e.getMessage(), e);
              errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage());
          }
  
          if (errMsg.isEmpty()) {
              etlResult.setSucceeded(true);
          } else {
              etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
          }
          return etlResult;
      }
  
      private static void appendCondition(List<String> params, DbMapping dbMapping, DataSource ds,
                                          StringBuilder sql) {
          if (params != null && params.size() == 1 && dbMapping.getEtlCondition() == null) {
              AtomicBoolean stExists = new AtomicBoolean(false);
              // 验证是否有SYS_TIME字段
              Util.sqlRS(ds, sql.toString(), rs -> {
                  try {
                      ResultSetMetaData rsmd = rs.getMetaData();
                      int cnt = rsmd.getColumnCount();
                      for (int i = 1; i <= cnt; i++) {
                          String columnName = rsmd.getColumnName(i);
                          if ("SYS_TIME".equalsIgnoreCase(columnName)) {
                              stExists.set(true);
                              break;
                          }
                      }
                  } catch (Exception e) {
                      // ignore
                  }
                  return null;
              });
              if (stExists.get()) {
                  sql.append(" WHERE SYS_TIME >= '").append(params.get(0)).append("' ");
              }
          } else if (dbMapping.getEtlCondition() != null && params != null && params.size() > 0) {
              String etlCondition = dbMapping.getEtlCondition();
              int size = params.size();
              for (int i = 0; i < size; i++) {
                  etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
              }
  
              sql.append(" ").append(etlCondition);
          }
      }
  
      /**
       * 执行导入
       */
      private static boolean executeSqlImport(DataSource srcDS, Connection targetDSConnection, String sql, DbMapping dbMapping,
                                              AtomicLong successCount, List<String> errMsg, boolean debug) {
          try {
              Map<String, String> columnsMap = new LinkedHashMap<>();
              Map<String, Integer> columnType = new LinkedHashMap<>();
  
  
              PhoenixSupportUtil.sqlRS(targetDSConnection, "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping) + " LIMIT 1 ", rs -> {
                  try {
  
                      ResultSetMetaData rsd = rs.getMetaData();
                      int columnCount = rsd.getColumnCount();
                      List<String> columns = new ArrayList<>();
                      List<String> excludeColumns = dbMapping.getExcludeColumns();
                      for (int i = 1; i <= columnCount; i++) {
                          String lower = rsd.getColumnName(i).toLowerCase();
                          if (!excludeColumns.contains(lower)) {
                              columnType.put(lower, rsd.getColumnType(i));
                              columns.add(lower);
                          }
                      }
                      columnsMap.putAll(SyncUtil.getColumnsMap(dbMapping, columns));
                      return true;
                  } catch (Exception e) {
                      logger.error(e.getMessage(), e);
                      return false;
                  }
              });
              Util.sqlRS(srcDS, sql, rs -> {
                  int idx = 1;
  
                  try {
                      boolean completed = false;
  
                      // if (dbMapping.isMapAll()) {
                      // columnsMap = dbMapping.getAllColumns();
                      // } else {
                      // columnsMap = dbMapping.getTargetColumns();
                      // }
  
                      StringBuilder insertSql = new StringBuilder();
                      insertSql.append("UPSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" (");
                      columnsMap
                              .forEach((targetColumnName, srcColumnName) -> insertSql.append(dbMapping.escape(targetColumnName)).append(","));
  
                      int len = insertSql.length();
                      insertSql.delete(len - 1, len).append(") VALUES (");
                      int mapLen = columnsMap.size();
                      for (int i = 0; i < mapLen; i++) {
                          insertSql.append("?,");
                      }
                      len = insertSql.length();
                      insertSql.delete(len - 1, len).append(")");
                      try (
                           //Connection connTarget = targetDS.getConnection();
                           Connection connTarget =PhoenixAdapter.getPhoenixConnection();
                           PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
                          connTarget.setAutoCommit(false);
  
                          while (rs.next()) {
                              completed = false;
  
                              pstmt.clearParameters();
  
                              // 删除数据
                              Map<String, Object> values = new LinkedHashMap<>();
                              StringBuilder deleteSql = new StringBuilder(
                                      "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE ");
                              appendCondition(dbMapping, deleteSql, values, rs);
                              try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
                                  int k = 1;
                                  for (Object val : values.values()) {
                                      pstmt2.setObject(k++, val);
                                  }
                                  pstmt2.execute();
                              }
  
                              Map<String, Object> insertValues = new HashMap<>();
                              int i = 1;
                              for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
                                  String targetColumnName = entry.getKey();
                                  String srcColumnName = entry.getValue();
                                  if (srcColumnName == null) {
                                      srcColumnName = targetColumnName;
                                  }
  
                                  Integer type = columnType.get(targetColumnName.toLowerCase());
  
                                  try {
                                      Object value = rs.getObject(srcColumnName);
                                      insertValues.put(srcColumnName, value);
                                      if (value != null) {
                                          SyncUtil.setPStmt(type, pstmt, value, i);
                                      } else {
                                          pstmt.setNull(i, type);
                                      }
                                  } catch (SQLException e) {
                                      insertValues.put(srcColumnName, null);
                                      pstmt.setNull(i, type);
                                  }
                                  i++;
                              }
                              if (debug) {
                                  logger.info("insert sql: {} {} {}", insertSql, insertValues, pstmt);
                              }
  
                              pstmt.execute();
                              if (logger.isTraceEnabled()) {
                                  logger.trace("Insert into target table, sql: {}", insertSql);
                              }
  
                              if (idx % dbMapping.getCommitBatch() == 0) {
                                  connTarget.commit();
                                  completed = true;
                              }
                              idx++;
                              successCount.incrementAndGet();
                              if (logger.isDebugEnabled()) {
                                  logger.debug("successful import count:" + successCount.get());
                              }
                          }
                          if (!completed) {
                              connTarget.commit();
                          }
                      }
  
                  } catch (Exception e) {
                      logger.error(dbMapping.getTable() + " etl failed! ==>" + e.getMessage(), e);
                      errMsg.add(dbMapping.getTable() + " etl failed! ==>" + e.getMessage());
                  }
                  return idx;
              });
              return true;
          } catch (Exception e) {
              logger.error(e.getMessage(), e);
              return false;
          }
      }
  
      /**
       * 拼接目标表主键where条件
       */
      private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map<String, Object> values,
                                          ResultSet rs) throws SQLException {
          // 拼接主键
          for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
              String targetColumnName = entry.getKey();
              String srcColumnName = entry.getValue();
              if (srcColumnName == null) {
                  srcColumnName = targetColumnName;
              }
              sql.append(dbMapping.escape(targetColumnName)).append("=? AND ");
              values.put(targetColumnName, rs.getObject(srcColumnName));
          }
          int len = sql.length();
          sql.delete(len - 4, len);
      }
  }
  