package com.product.data.center.service;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.product.core.cache.util.RedisUtil;
import com.product.core.entity.DataTableEntity;
import com.product.core.entity.FieldSetEntity;
import com.product.core.exception.BaseException;
import com.product.core.service.support.AbstractBaseService;
import com.product.core.spring.context.SpringMVCContextHolder;
import com.product.data.center.config.CmnConst;
import com.product.data.center.config.ErrorCode;
import com.product.data.center.entity.JournalEntity;
import com.product.data.center.utils.CustomLock;
import com.product.data.center.utils.QuerySqlParseUtil;
import com.product.data.center.utils.SqlTransferUtil;
import com.product.data.center.utils.WriteUtil;
import com.product.data.service.SyncDataConfigService;
import com.product.datasource.config.DataBaseType;
import com.product.datasource.dao.Dao;
import com.product.datasource.entity.DataBaseEntity;
import com.product.datasource.service.RedisService;
import com.product.util.BaseUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author cheng
 * @Date 2022/7/26 9:41
 * @Desc 数据归档
 */
@Service
public class DataArchivingService extends AbstractBaseService {

	@Autowired
	JournalManagerService journalManagerService;
	@Autowired
	DataArchivingQueue dataArchivingQueue;

	@Resource
	private SyncDataConfigService syncDataConfigService;

	private static CustomLock lock = new CustomLock();

	private Map<String, String> getCollectIds(String[] deleteDataSource, String tableName) {
		StringBuilder sql = new StringBuilder("SELECT ");
		sql.append("id,data_source FROM product_sys_data_collect WHERE ");
		List<Object> params = new ArrayList<>();
		for (int i = 0; i < deleteDataSource.length; i++) {
			String source = deleteDataSource[i];
			if (i > 0) {
				sql.append(" or ");
			}
			sql.append("( data_source = ? and LOWER(source_table) = ? )");
			params.add(source);
			params.add(tableName.toLowerCase());
		}
		DataTableEntity dataTableEntity = getBaseDao().listTable(sql.toString(), params.toArray());
		if (DataTableEntity.isEmpty(dataTableEntity)) {
			return null;
		}
		Map<String, String> collect = dataTableEntity.getData().stream().collect(Collectors.toMap(item -> item.getString("data_source"), item -> item.getString("id")));

		if (collect.size() != deleteDataSource.length) {
			throw new BaseException(ErrorCode.GET_COLLECT_ID_FAIL);
		}

		return collect;
	}

	public void reDeal(String logUuid) {
		FieldSetEntity fs = getBaseDao().getFieldSetEntity(CmnConst.PRODUCT_SYS_DATA_CENTER_LOG, logUuid, false);
		if (!FieldSetEntity.isEmpty(fs)) {
			String configUid = fs.getString(CmnConst.CONFIG_UUID);
			reDealArchiving(configUid);
		}
		fs.setValue(CmnConst.DEAL_FLAG, 1);
		getBaseDao().update(fs);
	}

	@Async
	public void reDealArchiving(String uuid) {
		this.dataArchivingEntry(uuid);
	}

	/**
	 * 扫码库删除
	 *
	 * @param configFse
	 * @throws BaseException
	 */
	private String sweepCodeLibrary(FieldSetEntity configFse) throws BaseException, SQLException {
		JournalEntity journalEntity = new JournalEntity();
		journalEntity.setDetail(3);
		TimeInterval timer = DateUtil.timer();
		int deleteSuccessCount = 0;
		try {
			String deleteDataSource = configFse.getString("delete_data_source");
			WriteUtil.append("DA-deleteDataSource：" + deleteDataSource);
			WriteUtil.append("DA-sub_delete_select_filter：" + configFse.getString("delete_select_filter"));
			if (StringUtils.isEmpty(deleteDataSource)) {
				return null;
			}
			String[] deleteDataSourceArray = deleteDataSource.split(",");
//            DataBaseEntity dbe = new DataBaseEntity(deleteDataSource);

			String source_data_validation = configFse.getString("source_data_validation");
			if (StringUtils.isEmpty(source_data_validation)) {
				return null;
			}
			String deleteDataTable = configFse.getString("delete_data_table");
			if (StringUtils.isEmpty(deleteDataTable)) {
				return null;
			}

			String collectIdField = configFse.getString("collect_id_field");

			DataBaseEntity validationDbe = new DataBaseEntity(source_data_validation);

			//扫码库删除唯一标识
			String deleteUniqueField = configFse.getString("delete_unique_field");

			//验证表
			String validationTableName = configFse.getString("table_data_validation");
			//验证唯一字段
			String validationUniqueField = configFse.getString("validation_unique_field");
			//忽略对比字段
			List<String> ignoreComparisonFields =
					!StringUtils.isEmpty(configFse.getString("ignore_comparison_field")) ?
							Lists.newArrayList(Arrays.asList(configFse.getString("ignore_comparison_field").split(","))) : null;
			int pageSize = 500;
			DataTableEntity list;
			Dao dao = null;
			int currentCount = 0;
			Map<String, String> collectIds = getCollectIds(deleteDataSource.split(","), deleteDataTable);
			try {
				String minID = null;
				String maxID = null;
				for (int j = 0; j < deleteDataSourceArray.length; j++) {
					if (dao != null) {
						dao.closeConnection();
					}
					dao = new DataBaseEntity(deleteDataSourceArray[j]).getDao();
					do {
//						String sql = "SELECT  * FROM (SELECT * from " + deleteDataTable + " where " + configFse.getString("delete_select_filter") + " order by " + deleteUniqueField + " ) A  ";
//						if (!Objects.isNull(minID) && !Objects.isNull(maxID)) {
//							sql += " where " + deleteUniqueField + " > " + maxID;
//						}
						String filter = configFse.getString("delete_select_filter");
						if (!Objects.isNull(minID) && !Objects.isNull(maxID)) {
							filter += "  and ( " + deleteUniqueField + " > " + maxID + ")";
						}
						list = dao.getList(deleteDataTable, filter, new Object[]{}, deleteUniqueField, 1, pageSize);
						WriteUtil.append("DA-删除子库数据-表名：" + deleteDataTable);
						if (DataTableEntity.isEmpty(list)) {
							break;
						}
						currentCount = list.getRows();
						Object[] uniqueValues = list.getData().stream().map(item -> item.getString(deleteUniqueField)).toArray();
						DataTableEntity validationData = validationDbe.getDao().getList(validationTableName, collectIdField + "= ? AND " +
								BaseUtil.buildQuestionMarkFilter(validationUniqueField, uniqueValues.length, true), ArrayUtil.addAll(new Object[]{collectIds.get(deleteDataSourceArray[j])}, uniqueValues));
						validationDbe.getDao().closeConnection();
						if (DataTableEntity.isEmpty(validationData)) {
							break;
						}
						Map<String, Map> collectMap = validationData.getData().stream().collect(Collectors.toMap(
								(item) -> item.getString(validationUniqueField),
								item -> item.getValues()));
						List<String> deleteUniqueValueList = new ArrayList<>();
						for (int i = 0; i < list.getRows(); i++) {
							FieldSetEntity fs = list.getFieldSetEntity(i);
							String uniqueValue = fs.getString(deleteUniqueField);
							Map<String, Object> map = collectMap.get(uniqueValue);
							if (ignoreComparisonFields != null && map != null) {
								for (int i1 = 0; i1 < ignoreComparisonFields.size(); i1++) {
									try {
										map.remove(ignoreComparisonFields.get(i1));
										fs.remove(ignoreComparisonFields.get(i1));
									} catch (Exception e) {
										e.printStackTrace();
									}
								}
							}
							if (map == null || !new JSONObject((Map) fs.getValues()).equals(new JSONObject(map))) {
								list.removeFieldSetEntity(i);
								i--;
								continue;
							}
							deleteUniqueValueList.add(uniqueValue);
							minID = getAimID(minID, uniqueValue, 0);
							maxID = getAimID(maxID, uniqueValue, 1);
						}
						collectMap.clear();
						if (deleteUniqueValueList.size() > 0) {
							deleteSuccessCount += dao.deleteRInt(configFse.getString("delete_data_table"),
									BaseUtil.buildQuestionMarkFilter(deleteUniqueField, deleteUniqueValueList.size(), true), deleteUniqueValueList.toArray());
							WriteUtil.append("DA-删除子库数据-已经删除条数：" + deleteSuccessCount);

						}
					} while (currentCount == pageSize);
				}
				journalEntity.setResult(1);
				journalEntity.setMin_id(minID);
				journalEntity.setMax_id(maxID);
			} catch (Exception e) {
				throw e;
			} finally {
				if (dao != null) {
					dao.closeConnection();
				}
			}
		} catch (Exception e) {
			SpringMVCContextHolder.getSystemLogger().error(e);
			journalEntity.setResult(0);
			journalEntity.setError(journalManagerService.getStackTrace(e));
			throw e;
		} finally {
			journalEntity.setCount(deleteSuccessCount);
			journalEntity.setCreated_utc_datetime(new Date());
			journalEntity.setSingle_duration(timer.intervalMs());
			journalEntity.setType(5);
			journalEntity.setConfigUuid(configFse.getUUID());
		}
		if (journalEntity != null && (journalEntity.getCount() > 0 || !StringUtils.isEmpty(journalEntity.getError()) || journalEntity.getResult() != 1)) {
			return journalManagerService.autoCreateJournal(journalEntity).getUUID();
		}
		return null;
	}

	/**
	 * 数据归档入口
	 *
	 * @param uuid 归档配置uuid
	 */
	public void dataArchivingEntry(String uuid) {
		if (lock.tryLock(uuid)) {
			try {
				dataArchivingEntryLock(uuid);
			} catch (Exception e) {
				throw e;
			} finally {
				lock.unLock(uuid);
			}
			return;
		}
		SpringMVCContextHolder.getSystemLogger().info("跳过执行：" + uuid);
	}

	public void dataArchivingEntryLock(String uuid) {
		SpringMVCContextHolder.getSystemLogger().info("开始执行归档》》》》" + uuid);
		WriteUtil.append("DA》》》");
		WriteUtil.append("DA-已经获取到锁");
		TimeInterval timer = DateUtil.timer();
		String curDateTimeStr = DateUtil.date().toString();
		int archivingSuccessCount = 0;
		JournalEntity journalEntity = new JournalEntity();
		journalEntity.setResult(1);
		try {
			//获取归档服务配置详情
			FieldSetEntity configFse = getBaseDao().getFieldSetEntity(CmnConst.DATA_ARCHIVING_TABLE, uuid, false);
			TimeInterval tempTestTimer = DateUtil.timer();
			//来源数据表
			String sourceTable = configFse.getString("source_table");
			boolean canExecuteClearFlag = canExecuteClear(sourceTable);
			String deleteSubLogUUID = null;

			if (canExecuteClearFlag) {
				WriteUtil.append("DA-删除扫码库》》》");
				//删除扫码库已提取到mes主库且根据配置条件过滤的数据 KT特有
//				deleteSubLogUUID = this.sweepCodeLibrary(configFse);
				//更改为异步执行
				ThreadUtil.execAsync(() -> {
					try {
						sweepCodeLibrary(configFse);
					} catch (SQLException e) {
						e.printStackTrace();
					}
				});
			}

			//来源数据源
			String sourceDataSource = configFse.getString("source_data_source");
			DataBaseEntity sourceDbe = new DataBaseEntity(sourceDataSource);
			//目标数据源
			String targetDataSource = configFse.getString("target_data_source");
			DataBaseEntity targetDbe = new DataBaseEntity(targetDataSource);
			//目标数据表前缀
			String targetTablePrefix = configFse.getString("target_table_prefix");
			if (targetTablePrefix.endsWith("_")) {
				targetTablePrefix = targetTablePrefix.substring(0, targetTablePrefix.length() - 1);
			}
			//来源采集ID字段
			String sourceCollectIdField = configFse.getString("source_collect_id_field");
			//来源唯一标识字段 由MES子库生成的值
			String sourceUniqueField = configFse.getString("source_unique_field");
			//唯一字段
			String uniqueField = configFse.getString("unique_field");
			// 时间字段
			String timeField = configFse.getString("time_field");
			// 提取时间字段
			String extractTimeField = configFse.getString(CmnConst.EXTRACT_TIME_FIELD);

			Dao sourceDao = sourceDbe.getDao();
			Dao targetDao = targetDbe.getDao();
			Set<String> targetTableSet;
			try {
				DataArchivingServiceImpl service = new DataArchivingServiceImpl(sourceDao, targetDao, sourceTable, uuid, sourceDbe.getDbName(), targetDbe.getDbName());
				String keyPrefix = "DA_STORE:" + sourceTable + ":";
				boolean serializeFlag = true;
				int outTime = 60 * 60;// 60 min
				RedisService readRedis = StringUtils.isEmpty(configFse.getString("redis_config_uuid")) ? new RedisService() : new RedisService(new DataBaseEntity(configFse.getString("redis_config_uuid")));
				tempTestTimer = DateUtil.timer();
				WriteUtil.append("DA-获取表名集合》》》");
				if (!StringUtils.isEmpty(timeField)) {
					targetTableSet = QuerySqlParseUtil.getAllTableName(targetDao, targetDbe.getDbName(), sourceTable);
				} else {
					targetTableSet = Sets.newLinkedHashSet();
					targetTableSet.add(targetTablePrefix);
				}
				WriteUtil.append("DA-获取表名集合耗时：" + tempTestTimer.intervalMs());
				FieldSetEntity paramFse = getBaseDao().getFieldSetBySQL("select max(statistics_final_time) statistics_final_time from product_sys_data_center_log where type=5 and detail=4 and result=1 and config_uuid=?", new Object[]{uuid}, false);
				Date preMaxTime = paramFse.getDate("statistics_final_time");
				// 没有成功的最大统计时间：要么是第一次归档，要么是归档的表没有设定时间字段（所有数据都在一张表上，每次都会拉取这张表当时的所有内容）
				boolean turnRedisFilterFlag = preMaxTime == null;
				boolean turnTargetDBClearFlag = false;
				if (turnRedisFilterFlag) {
					// 若是没有成功的日志，表示为第一次归档，跳过目标库数据清理
					FieldSetEntity logExistsFse = getBaseDao().getFieldSetBySQL("select count(1) count_value from product_sys_data_center_log where type=5 and detail=4 and result=1 and config_uuid=?", new Object[]{uuid}, false);
					turnTargetDBClearFlag = "0".equals(logExistsFse.getString("count_value"));
				}
				StringBuilder paramSql = new StringBuilder(128);
				paramSql.append("select max(").append(uniqueField).append(") max_id,min(").append(uniqueField).append(") min_id");
				if (!StringUtils.isEmpty(timeField)) {
					paramSql.append(",max(").append(timeField).append(") max_update_time,min(").append(timeField).append(") min_update_time");
				}
				if (!StringUtils.isEmpty(extractTimeField)) {
					paramSql.append(",max(").append(extractTimeField).append(") max_extract_time,min(").append(extractTimeField).append(") min_extract_time");
				}
				paramSql.append(" from ").append(sourceTable);
				StringBuilder filterSb = new StringBuilder(128);
				if (preMaxTime != null && (!StringUtils.isEmpty(timeField) || !StringUtils.isEmpty(extractTimeField))) {
					int sourceDbType = sourceDbe.getDbType().getValue();
					if (!StringUtils.isEmpty(timeField)) {
						filterSb.append("(").append(timeField).append(">=").append(SqlTransferUtil.addDate(sourceDbType, SqlTransferUtil.str2Date(sourceDbType, preMaxTime), -2, SqlTransferUtil.MINUTE))
								.append(" and ").append(timeField).append("<=").append(SqlTransferUtil.str2Date(sourceDbType, curDateTimeStr)).append(")");
					}
					if (!StringUtils.isEmpty(extractTimeField)) {
						if (filterSb.length() > 0) {
							filterSb.append(" or ");
						}
						filterSb.append("(").append(extractTimeField).append(">=").append(SqlTransferUtil.addDate(sourceDbType, SqlTransferUtil.str2Date(sourceDbType, preMaxTime), -2, SqlTransferUtil.MINUTE))
								.append(" and ").append(extractTimeField).append("<=").append(SqlTransferUtil.str2Date(sourceDbType, curDateTimeStr)).append(")");
					}
					paramSql.append(" where (").append(filterSb).append(")");
				}
				paramFse = sourceDao.getOne(paramSql.toString());
				if (filterSb.length() > 0) {
					filterSb.insert(0, "(");
					filterSb.append(") and ");
				}
				if (StringUtils.isEmpty(paramFse.getString("max_id"))) {
					WriteUtil.append("最大id为空，跳出，执行sql：" + paramSql);
					return;
				}
				filterSb.append(uniqueField).append("<='").append(paramFse.getString("max_id")).append("'");
				if (!StringUtils.isEmpty(configFse.getString("select_filter")) && !StringUtils.isEmpty(configFse.getString("select_filter").trim())) {
					filterSb.append(" and (").append(configFse.getString("select_filter")).append(")");
				}
				WriteUtil.append("DA-sql-filter：" + filterSb);
				Date statisticsStartTime = getAimDate(paramFse.getDate("min_update_time"), paramFse.getDate("min_extract_time"), 0);
				Date statisticsFinalTime = getAimDate(paramFse.getDate("max_update_time"), paramFse.getDate("max_extract_time"), 0);
				String maxID = paramFse.getString("max_id");
				String minID = paramFse.getString("min_id");
				String splitTableType = "1".equals(configFse.getString("split_table_type")) ? "1" : "0";
				//开启队列查询时，先清空队列防止上次未处理完的数据占用内存
				dataArchivingQueue.clear(sourceTable);
				dataArchivingQueue.query(sourceDbe, sourceTable, filterSb.toString(), null, uniqueField, minID);
				DataTableEntity allDte;
				Map<String, List<DataTableEntity>> groupDteMap;
				do {
					allDte = dataArchivingQueue.get(sourceTable);
					if (DataTableEntity.isEmpty(allDte)) {
						WriteUtil.append("DA-从队列中获取内容为空，执行睡眠跳过...");
						Thread.sleep(RandomUtil.randomInt(800, 1200));
						continue;
					}
					WriteUtil.append("DA-从队列中获取内容非空，执行插入...");
					groupDteMap = dteGroupByTime(allDte, timeField, splitTableType);
					FieldSetEntity tempFse;
					String field;
					Object value;
					Date date;
					Date updateTime;
					Date extractTime;
					Map<String, Set<String>> sourceTableUniqueByCollectId = null;
					for (Map.Entry<String, List<DataTableEntity>> entry : groupDteMap.entrySet()) {
						String time = entry.getKey();
						List<DataTableEntity> dteList = entry.getValue();
						for (DataTableEntity list : dteList) {
							//创建表（不存在才创建）
							tempTestTimer = DateUtil.timer();
							WriteUtil.append("DA-创建表》》》");
							String tableName = service.createTable(targetTablePrefix, time, targetDbe);
							WriteUtil.append("DA-创建表耗时：" + tempTestTimer.intervalMs());
							targetTableSet.add(tableName);
							FieldSetEntity fs = list.getFieldSetEntity(0);
							tempTestTimer = DateUtil.timer();
							WriteUtil.append("DA-数据筛选》》》");
							if (!StringUtils.isEmpty(sourceCollectIdField)) {
								sourceTableUniqueByCollectId = new HashMap<>();
							}
							// 没有成功的最大统计时间：要么是第一次归档，要么是归档的表没有设定时间字段（所有数据都在一张表上，每次都会拉取这张表当时的所有内容），跳过验证
							if (!turnRedisFilterFlag) {
								// 验证是否存在redis中，若是存在，比较时间字段值的大小，若是查询出数据中的时间更靠近当前时间，那么重置redis中的时间和过期时间，若是redis中的时间更靠近当前时间，则剔除数据集中的数据，并重置过期时间；若是不存在，则正常执行，先删除然后新增
								for (int i = list.getRows() - 1; i >= 0; i--) {
									tempFse = list.getFieldSetEntity(i);
									updateTime = tempFse.getDate(timeField);
									extractTime = tempFse.getDate(extractTimeField);
									date = getAimDate(updateTime, extractTime, 0);
									field = tempFse.getString(uniqueField);
									value = readRedis.get(keyPrefix + field, serializeFlag);
									String sourceUniqueValue = tempFse.getString(sourceUniqueField);
									String collectId = tempFse.getString(sourceCollectIdField);
									if (!StringUtils.isEmpty(sourceCollectIdField) && sourceTableUniqueByCollectId != null
											&& !StringUtils.isEmpty(collectId) && !StringUtils.isEmpty(sourceUniqueValue)) {
										Set<String> uniqueValues = sourceTableUniqueByCollectId.computeIfAbsent(collectId, k -> new HashSet<>());
										uniqueValues.add(sourceUniqueValue);
									}
									if (value != null && !StringUtils.isEmpty(value.toString())) {
										if (((Date) value).compareTo(date) <= 0) {
											readRedis.set(keyPrefix + field, date, serializeFlag);
											readRedis.setOutTime(keyPrefix + field, outTime, serializeFlag);
											statisticsStartTime = getAimDate(statisticsStartTime, date, 0);
											statisticsFinalTime = getAimDate(statisticsFinalTime, date, 1);
											minID = getAimID(minID, field, 0);
											maxID = getAimID(maxID, field, 1);
										} else {
											list.removeFieldSetEntity(i);
											readRedis.setOutTime(keyPrefix + field, outTime, serializeFlag);
										}
									} else {
										readRedis.set(keyPrefix + field, date, serializeFlag);
										readRedis.setOutTime(keyPrefix + field, outTime, serializeFlag);
										statisticsStartTime = getAimDate(statisticsStartTime, date, 0);
										statisticsFinalTime = getAimDate(statisticsFinalTime, date, 1);
										minID = getAimID(minID, field, 0);
										maxID = getAimID(maxID, field, 1);
									}
								}
							}
							WriteUtil.append("DA-数据筛选耗时：" + tempTestTimer.intervalMs());
							if (DataTableEntity.isEmpty(list)) {
								continue;
							}
							//重设表名
							fs.setTableName(tableName);
							list.setMeta(fs.getMeta());
							Connection connection = sourceDao.getConnection();
							//设置手动提交
							connection.setAutoCommit(false);
							try {
								tempTestTimer = DateUtil.timer();
								WriteUtil.append("DA-清理数据》》》");
								// 拉取全量数据到空表（第一次归档），跳过清理
								if (!turnTargetDBClearFlag) {
									// 清理数据
									DataTableEntity clearDte = clearArchiveRepeatData(sourceTable, targetTableSet, uniqueField, targetDao, list);
									WriteUtil.append("DA-清理数据量：" + clearDte.getRows());
								}
								WriteUtil.append("DA-清理数据耗时：" + tempTestTimer.intervalMs());
								// 新增数据
								tempTestTimer = DateUtil.timer();
								WriteUtil.append("DA-新增数据量：" + list.getRows());
								if (list.getRows() > 0) {
									WriteUtil.append("DA-新增数据》》》表名：" + list.getFieldSetEntity(0).getMeta().getTableName()[0]);
									try {
										targetDao.addBatch(list);
									} catch (Exception e) {
										if (turnTargetDBClearFlag) {
											clearArchiveRepeatData(sourceTable, targetTableSet, uniqueField, targetDao, list);
											targetDao.addBatch(list);
										} else {
											throw e;
										}
									}
								}
								WriteUtil.append("DA-新增数据耗时：" + tempTestTimer.intervalMs());
								// 提交
								connection.commit();
							} catch (Exception e) {

								//若批量添加失败回滚删除
								try {
									connection.rollback();
								} catch (Exception er) {
									e.printStackTrace();
									throw er;
								}
								throw e;
							} finally {
								//重设连接为自动提交
								connection.setAutoCommit(true);
							}
							archivingSuccessCount += list.getRows();
						}
					}
				} while (!dataArchivingQueue.checkQueryFinish(sourceTable) || !dataArchivingQueue.checkInsertQueueEmpty(sourceTable));
				journalEntity.setSingle_duration(timer.intervalMs());
				journalEntity.setStatistics_start_time(statisticsStartTime);
				journalEntity.setStatistics_final_time(statisticsFinalTime);
				journalEntity.setMin_id(minID);
				journalEntity.setMax_id(maxID);
				WriteUtil.append("DA-循环完毕");
			} catch (Exception e) {
				WriteUtil.append("error:\n" + journalManagerService.getStackTrace(e));
				throw e;
			} finally {
				targetDao.closeConnection();
				sourceDao.closeConnection();
				// 关闭线程
				dataArchivingQueue.shutdownQueryThread(sourceTable);
			}

			// 删除mes主库的内容
			if (canExecuteClearFlag) {
				WriteUtil.append("DA-删除mes主库内容》》》");
				String deleteMasterLogUUID = deleteMasterData(sourceTable, configFse, sourceDao, targetDao, uniqueField, timeField, extractTimeField, targetTableSet, deleteSubLogUUID);
				if (!StringUtils.isEmpty(deleteMasterLogUUID)) {
					journalEntity.setPre_step_uuid(deleteMasterLogUUID);
				} else if (!StringUtils.isEmpty(deleteSubLogUUID)) {
					journalEntity.setPre_step_uuid(deleteMasterLogUUID);
				}
				WriteUtil.append("DA-删除mes主库内容完毕");
			}
			String errorInfo = dataArchivingQueue.getErrorLog(sourceTable);
			if (!StringUtils.isEmpty(errorInfo)) {
				journalEntity.setError(errorInfo);
				journalEntity.setResult(0);
			}
		} catch (Exception e) {
			SpringMVCContextHolder.getSystemLogger().error(e);
			if (!StringUtils.isEmpty(journalEntity.getError())) {
				journalEntity.setError(journalEntity.getError() + "\n" + journalManagerService.getStackTrace(e));
			} else {
				journalEntity.setError(journalManagerService.getStackTrace(e));
			}
			journalEntity.setResult(0);
		} finally {
			journalEntity.setSingle_duration(timer.intervalMs());
			journalEntity.setCount(archivingSuccessCount);
			journalEntity.setConfigUuid(uuid);
			journalEntity.setType(5);
			journalEntity.setDetail(4);
			journalEntity.setCreated_utc_datetime(new Date());
			FieldSetEntity curLogFse = null;
			if (archivingSuccessCount > 0 || !StringUtils.isEmpty(journalEntity.getError()) || journalEntity.getResult() != 1 || !StringUtils.isEmpty(journalEntity.getPre_step_uuid())) {
				curLogFse = journalManagerService.autoCreateJournal(journalEntity);
			}

			// 将日志表中执行失败的记录标记为已经重新处理
			WriteUtil.append("DA-将日志表中执行失败的记录标记为已经重新处理");
			DataTableEntity logDte;
			if (curLogFse == null) {
				logDte = getBaseDao().listTable(CmnConst.PRODUCT_SYS_DATA_CENTER_LOG, "type=5 and result=0 and deal_flag=0 and config_uuid=?", new Object[]{uuid});
			} else {
				logDte = getBaseDao().listTable(CmnConst.PRODUCT_SYS_DATA_CENTER_LOG, "type=5 and result=0 and deal_flag=0 and config_uuid=? and uuid<>?", new Object[]{uuid, curLogFse.getUUID()});
			}
			for (int i = 0; i < logDte.getRows(); i++) {
				journalManagerService.writeBackReDealResult(logDte.getFieldSetEntity(i), true);
			}
			WriteUtil.append("DA-执行完毕");
		}
	}

	/**
	 * 清理归档重复数据
	 *
	 * @param sourceTable
	 * @param targetTableSet
	 * @param uniqueField
	 * @param targetDao
	 */
	private DataTableEntity clearArchiveRepeatData(String sourceTable, Set<String> targetTableSet, String uniqueField, Dao targetDao, DataTableEntity list) {
		StringBuilder clearSql = new StringBuilder(128);
		clearSql.append("with ").append(QuerySqlParseUtil.getUnionTableSql(sourceTable, targetTableSet, Arrays.asList(uniqueField, "{#table_name#}"), false, " where " + BaseUtil.buildQuestionMarkFilter(uniqueField, list.getFieldAllValues(uniqueField), true)))
				.append("\nselect ").append(uniqueField).append(",_table_name from ").append(sourceTable);
		DataTableEntity clearDte = targetDao.getList(clearSql.toString());
		if (!DataTableEntity.isEmpty(clearDte)) {
			Set<Object> clearTableSet = Sets.newHashSet();
			clearTableSet.addAll(Arrays.asList(clearDte.getFieldAllValues("_table_name")));
			for (Object targetTableName : clearTableSet) {
				targetDao.delete(targetTableName.toString(),
						BaseUtil.buildQuestionMarkFilter(uniqueField, clearDte.getRows(), true),
						clearDte.getData().stream().map(item -> item.getString(uniqueField)).toArray());
			}
		}
		return clearDte;
	}

	/**
	 * 清理判定，每天切换的时候可以执行一次
	 *
	 * @param tableName
	 * @return
	 */
	public boolean canExecuteClear(String tableName) {
		String dateStr = DateUtil.format(DateUtil.date(), "yyyy-MM-dd");
		final String KEY = "DE_CLEAR_STORE";
		Object preDate = RedisUtil.getHash(KEY, tableName);
		if (dateStr.equals(preDate)) {
			return false;
		} else {
			RedisUtil.setHash(KEY, tableName, dateStr);
			return true;
		}
	}

	/**
	 * 清理主库数据
	 *
	 * @param sourceTable      源表名
	 * @param configFse        配置fse
	 * @param sourceDao        源dao
	 * @param targetDao        目标dao
	 * @param uniqueField      唯一字段
	 * @param timeField        更新（归档）时间字段
	 * @param extractTimeField 提取时间字段
	 * @param targetTableSet   目标表集合
	 * @param deleteSubLogUUID 删除子表日志uuid
	 * @return
	 */
	private String deleteMasterData(String sourceTable, FieldSetEntity configFse, Dao sourceDao, Dao targetDao, String uniqueField, String timeField, String extractTimeField, Set<String> targetTableSet, String deleteSubLogUUID) {
		FieldSetEntity deleteMasterLogFse = new FieldSetEntity();
		deleteMasterLogFse.setTableName(CmnConst.PRODUCT_SYS_DATA_CENTER_LOG);
		TimeInterval deleteMasterLogTimer = DateUtil.timer();
		String minID = "";
		String maxID = "";
		Date statisticsStartTime = null;
		Date statisticsFinalTime = null;
		try {
			StringBuilder sql = new StringBuilder(128);
			sql.append("select count(*) count_value from ").append(sourceTable);
			if (!StringUtils.isEmpty(configFse.getString("source_select_filter"))) {
				sql.append(" where ").append(configFse.getString("source_select_filter"));
			} else {
				sql.append(" where  1=2");
			}
			FieldSetEntity paramFse = sourceDao.getOne(sql.toString());
			WriteUtil.append("DA-sourceDao.getOne(sql.toString())-sql:" + sql);
			int totalCount = StringUtils.isEmpty(paramFse.getString("count_value")) ? 0 : paramFse.getInteger("count_value");
			int delTotalCount = 0;
			if (totalCount > 0) {
				int pageSize = 1000;
				int totalPage = totalCount / pageSize + (totalCount % pageSize == 0 ? 0 : 1);
				sql.setLength(0);
				sql.append("select ").append(uniqueField).append(" from ").append(sourceTable);
				if (!StringUtils.isEmpty(configFse.getString("source_select_filter"))) {
					sql.append(" where ").append(configFse.getString("source_select_filter"));
				}
				StringBuilder existSql = new StringBuilder(128);
				existSql.append("with ").append(QuerySqlParseUtil.getUnionTableSql(sourceTable, targetTableSet))
						.append("\nselect ").append(uniqueField).append(" from ").append(sourceTable);
				StringBuilder tempSql = new StringBuilder(128);
				for (int i = 0; i < totalPage; i++) {
					DataTableEntity delDte = sourceDao.getList(sql.toString(), new Object[]{}, 1, pageSize);
					WriteUtil.append("DA-删除的数据-delDte：" + Arrays.toString(delDte.getFieldAllValues(uniqueField)));
					if (DataTableEntity.isEmpty(delDte)) {
						continue;
					}
					// 验证归档库里面存在，仅删除存在，不存在的保留
					tempSql.setLength(0);
					tempSql.append(existSql);
					tempSql.append(" where ").append(BaseUtil.buildQuestionMarkFilter(uniqueField, delDte.getFieldAllValues(uniqueField), true));
					WriteUtil.append("DA-tempSql:" + tempSql);
					DataTableEntity existDte = targetDao.getList(tempSql.toString(), new Object[]{});
					if (existDte.getRows() > 0) {
						FieldSetEntity existFse;
						for (int j = 0; j < existDte.getRows(); j++) {
							existFse = existDte.getFieldSetEntity(j);
							Date updateTime = existFse.getDate(timeField);
							Date extractTime = existFse.getDate(extractTimeField);
							Date date = getAimDate(updateTime, extractTime, 0);
							statisticsStartTime = getAimDate(statisticsStartTime, date, 0);
							statisticsFinalTime = getAimDate(statisticsFinalTime, date, 1);
							minID = getAimID(minID, existFse.getString(uniqueField), 0);
							maxID = getAimID(maxID, existFse.getString(uniqueField), 1);
						}
						delTotalCount += existDte.getRows();
						sourceDao.delete(sourceTable,
								BaseUtil.buildQuestionMarkFilter(uniqueField, existDte.getRows(), true),
								existDte.getData().stream().map(item -> item.getString(uniqueField)).toArray());
					}
				}
				WriteUtil.append("DA-删除总条数:" + delTotalCount);
			}
			deleteMasterLogFse.setValue(CmnConst.COUNT, delTotalCount);
			deleteMasterLogFse.setValue(CmnConst.RESULT, 1);
		} catch (Exception e) {
			e.printStackTrace();
			deleteMasterLogFse.setValue(CmnConst.RESULT, 0);
			deleteMasterLogFse.setValue(CmnConst.ERROR, journalManagerService.getStackTrace(e));
		} finally {
			targetDao.closeConnection();
			sourceDao.closeConnection();
			deleteMasterLogFse.setValue(CmnConst.CREATED_UTC_DATETIME, new Date());
			deleteMasterLogFse.setValue(CmnConst.TYPE, 5);
			deleteMasterLogFse.setValue(CmnConst.DETAIL, 6);
			deleteMasterLogFse.setValue(CmnConst.PRE_STEP_UUID, deleteSubLogUUID);
			deleteMasterLogFse.setValue(CmnConst.DEAL_FLAG, 0);
			deleteMasterLogFse.setValue(CmnConst.DEAL_RESULT, 1);
			deleteMasterLogFse.setValue(CmnConst.MIN_ID, minID);
			deleteMasterLogFse.setValue(CmnConst.MAX_ID, maxID);
			deleteMasterLogFse.setValue(CmnConst.SINGLE_DURATION, deleteMasterLogTimer.intervalMs());
			deleteMasterLogFse.setValue(CmnConst.CONFIG_UUID, configFse.getUUID());
			deleteMasterLogFse.setValue(CmnConst.STATISTICS_START_TIME, statisticsStartTime);
			deleteMasterLogFse.setValue(CmnConst.STATISTICS_FINAL_TIME, statisticsFinalTime);
			if ((!StringUtils.isEmpty(deleteMasterLogFse.getString(CmnConst.COUNT)) && deleteMasterLogFse.getInteger(CmnConst.COUNT) > 0)
					|| !StringUtils.isEmpty(deleteMasterLogFse.getString(CmnConst.ERROR)) || !"1".equals(deleteMasterLogFse.getString(CmnConst.RESULT))) {
				getBaseDao().add(deleteMasterLogFse);
			}
		}
		return deleteMasterLogFse.getUUID();
	}

	/**
	 * 获取指定的日期，若有一个为空，那么直接获取另外一个的值；否则按照指定取值
	 *
	 * @param d1
	 * @param d2
	 * @param sign 大于0，取两者中大的，就是更靠近当前时间的；否则取小的，就是更远离当前时间的
	 * @return
	 */
	private Date getAimDate(Date d1, Date d2, int sign) {
		if (d1 == null && d2 == null) {
			return null;
		}
		if (d1 == null || d2 == null) {
			return d1 == null ? d2 : d1;
		}
		if (sign > 0) {
			return d1.compareTo(d2) > 0 ? d1 : d2;
		} else {
			return d1.compareTo(d2) > 0 ? d2 : d1;
		}
	}

	/**
	 * 获取指定的ID，那么直接获取另外一个的值；否则按照指定取值
	 *
	 * @param s1
	 * @param s2
	 * @param sign 大于0，取两者中大的；否则取小的
	 * @return
	 */
	private String getAimID(String s1, String s2, int sign) {
		if (StringUtils.isEmpty(s1) && StringUtils.isEmpty(s2)) {
			return null;
		}
		if (StringUtils.isEmpty(s1) || StringUtils.isEmpty(s2)) {
			return StringUtils.isEmpty(s1) ? s2 : s1;
		}
		String numberRegexp = "\\d{1,11}";
		if (s1.matches(numberRegexp) && s2.matches(numberRegexp)) {
			boolean b = Long.parseLong(s1) > Long.parseLong(s2);
			if (sign > 0) {
				return b ? s1 : s2;
			} else {
				return b ? s2 : s1;
			}
		} else {
			if (sign > 0) {
				return s1.compareTo(s2) > 0 ? s1 : s2;
			} else {
				return s1.compareTo(s2) > 0 ? s2 : s1;
			}
		}
	}

	/**
	 * 将dte按照年份分组
	 *
	 * @param dte
	 * @param timeField      时间字段
	 * @param splitTableType 分表方式，0-年，1-月
	 * @return Map<时间, List < dte数据>>
	 */
	private Map<String, List<DataTableEntity>> dteGroupByTime(DataTableEntity dte, String timeField, String splitTableType) {
		Map<String, List<DataTableEntity>> groupDteMap = Maps.newHashMap();
		FieldSetEntity fse;
		String time;
		List<DataTableEntity> groupDteList;
		if (StringUtils.isEmpty(timeField)) {
			groupDteList = Lists.newArrayList();
			groupDte(groupDteList, dte);
			groupDteMap.put("0", groupDteList);
		} else {
			SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMM");
			for (int i = 0; i < dte.getRows(); i++) {
				fse = dte.getFieldSetEntity(i);
				if (fse.getDate(timeField) == null) {
					throw new BaseException(ErrorCode.DATA_ARCHIVE_FAIL_NO_SPLIT_TABLE_DATE.getValue(), ErrorCode.DATA_ARCHIVE_FAIL_NO_SPLIT_TABLE_DATE.getText() + "\ntable_name:" + fse.getTableName() + "\ndata:" + fse.getValues());
				}
				if ("1".equals(splitTableType)) {
					time = dateFormat.format(fse.getDate(timeField));
				} else {
					time = String.valueOf(DateUtil.year(fse.getDate(timeField)));
				}
				groupDteList = groupDteMap.computeIfAbsent(time, k -> Lists.newArrayList());
				groupAddDte(groupDteList, fse);
			}
		}
		return groupDteMap;
	}

	private void groupDte(List<DataTableEntity> list, DataTableEntity allDte) {
		if (list == null) {
			throw new BaseException(ErrorCode.DATA_ARCHIVE_GROUP_CONTAINER_IS_NULL);
		}
		for (int i = 0; i < allDte.getRows(); i++) {
			groupAddDte(list, allDte.getFieldSetEntity(i));
		}
	}

	private void groupAddDte(List<DataTableEntity> list, FieldSetEntity fse) {
		DataTableEntity dte;
		if (list.isEmpty()) {
			dte = new DataTableEntity();
			list.add(dte);
		} else {
			dte = list.get(list.size() - 1);
			if (dte.getRows() >= DataArchivingQueue.INSERT_PAGE_SIZE) {
				dte = new DataTableEntity();
				list.add(dte);
			}
		}
		dte.addFieldSetEntity(fse);
	}


	class DataArchivingServiceImpl {

		private Dao sourceDao;
		private Dao targetDao;
		private String sourceTable;
		private String configUid;
		private String sourceDbName;
		private String targetDbName;

		public DataArchivingServiceImpl(Dao sourceDao, Dao targetDao, String sourceTable, String configUid, String sourceDbName, String targetDbName) {
			this.sourceDao = sourceDao;
			this.targetDao = targetDao;
			this.sourceTable = sourceTable;
			this.configUid = configUid;
			this.sourceDbName = sourceDbName;
			this.targetDbName = targetDbName;
		}

		/**
		 * 验证数据表是否存在
		 *
		 * @param tableName 数据表名
		 * @return
		 */
		public boolean dataTableIsExists(String tableName) throws BaseException {
			try {
				Connection connection = targetDao.getConnection();
				DatabaseMetaData metaData = targetDao.getConnection().getMetaData();
				ResultSet tables = metaData.getTables(null, null, tableName, new String[]{"TABLE"});
				boolean result = tables.next();
				tables.close();
				connection.close();
				return result;
			} catch (Exception e) {
				e.printStackTrace();
				throw new BaseException(e);
			}
		}


		/**
		 * 创建表
		 *
		 * @param prefix 表名前缀
		 */
		public String createTable(String prefix, String time, DataBaseEntity dbe) throws BaseException {
			String tableName;
			if (!"0".equals(time)) {
				tableName = prefix + (prefix.lastIndexOf("_") != prefix.length() - 1 ? "_" : "") + time;
			} else {
				tableName = prefix.endsWith("_") ? prefix.substring(0, prefix.length() - 1) : prefix;
			}
			if (dataTableIsExists(tableName)) {
				return tableName;
			}
			JSONObject tableInfoObj = getTableInfo(tableName);
			String sql = getSql(tableInfoObj);
			SpringMVCContextHolder.getSystemLogger().error("sql:\n" + sql);
			//先创建记录再执行ddl语句不然报错后ddl无法回滚
			saveCreateTableRecord(tableName, time);
			targetDao.executeSql(sql);
//            syncDataConfigService.addTableField(dbe,dbe.getUuid(),tableName);
			return tableName;
		}

		/**
		 * 根据表名获取来源数据源对应表的结构信息
		 *
		 * @param tableName
		 * @return
		 */
		private JSONObject getTableInfo(String tableName) {
			JSONObject tableInfoObj = new JSONObject();
			tableInfoObj.put(CmnConst.NAME, tableName);
			DataTableEntity tempDte;
			FieldSetEntity tempFse;
			JSONObject fieldInfoObj = new JSONObject(new LinkedHashMap<>());
			tableInfoObj.put(CmnConst.FIELD, fieldInfoObj);
			JSONObject indexInfoObj = new JSONObject(new LinkedHashMap<>());
			tableInfoObj.put(CmnConst.INDEX, indexInfoObj);
			JSONObject singleFieldInfoObj;
			if (DataBaseType.MYSQL.equals(sourceDao.getDataBaseType())) {
				// mysql
				// 表
				tempFse = sourceDao.getOne("information_schema.`TABLES`", "table_schema=? and table_name=?", new Object[]{sourceDbName, sourceTable});
				tableInfoObj.put(CmnConst.COMMENT, tempFse.getString("table_comment"));
				// 字段表
				tempDte = sourceDao.getList("information_schema.`COLUMNS`", "table_schema=? and table_name=?", new Object[]{sourceDbName, sourceTable}, "ordinal_position", 1, Integer.MAX_VALUE);
				for (int i = 0; i < tempDte.getRows(); i++) {
					tempFse = tempDte.getFieldSetEntity(i);
					singleFieldInfoObj = new JSONObject();
					fieldInfoObj.put(tempFse.getString("column_name"), singleFieldInfoObj);
					singleFieldInfoObj.put(CmnConst.TYPE, tempFse.getString("data_type"));
					singleFieldInfoObj.put(CmnConst.INTEGER, StringUtils.isEmpty(tempFse.getString("character_maximum_length")) ? tempFse.getString("numeric_precision") : tempFse.getString("character_maximum_length"));
					singleFieldInfoObj.put(CmnConst.DECIMAL, tempFse.getString("numeric_scale"));
					singleFieldInfoObj.put(CmnConst.NULLABLE, "NO".equalsIgnoreCase(tempFse.getString("is_nullable")) ? 0 : 1);
					singleFieldInfoObj.put(CmnConst.COMMENT, tempFse.getString("column_comment"));
				}
				// 索引表
				StringBuilder sql = new StringBuilder(128);
				sql.append("select index_name,non_unique,group_concat(column_name) column_name");
				sql.append("\nfrom information_schema.`STATISTICS`");
				sql.append("\nwhere table_schema=? and table_name=?");
				sql.append("\ngroup by index_name,non_unique");
				tempDte = sourceDao.getList(sql.toString(), new Object[]{sourceDbName, sourceTable});
				for (int i = 0; i < tempDte.getRows(); i++) {
					tempFse = tempDte.getFieldSetEntity(i);
					singleFieldInfoObj = new JSONObject();
					indexInfoObj.put(tempFse.getString("index_name"), singleFieldInfoObj);
					singleFieldInfoObj.put(CmnConst.TYPE, "PRIMARY".equalsIgnoreCase(tempFse.getString("index_name")) ? CmnConst.PRIMARY : ("1".equals(tempFse.getString("non_unique")) ? CmnConst.NORMAL : CmnConst.UNIQUE));
					singleFieldInfoObj.put(CmnConst.COLUMN_NAME, tempFse.getString("column_name"));
				}
			} else if (DataBaseType.ORACLE.equals(sourceDao.getDataBaseType())) {
				// oracle
				// 表
				String upperTableName = sourceTable.toUpperCase();
				tempFse = sourceDao.getOne("SYS.USER_TAB_COMMENTS", "table_name=?", new Object[]{upperTableName});
				tableInfoObj.put(CmnConst.COMMENT, tempFse.getString("comments"));
				// 字段表
				StringBuilder sql = new StringBuilder(128);
				sql.append("SELECT TC.COLUMN_NAME,DATA_TYPE,DATA_LENGTH,DATA_PRECISION,DATA_SCALE,NULLABLE,CHAR_LENGTH,COMMENTS FROM SYS.USER_TAB_COLUMNS TC");
				sql.append("\nLEFT JOIN USER_COL_COMMENTS CC ON TC.TABLE_NAME=CC.TABLE_NAME AND TC.COLUMN_NAME=CC.COLUMN_NAME");
				sql.append("\nWHERE TC.TABLE_NAME=?");
				sql.append("\nORDER BY TC.COLUMN_ID");
				tempDte = sourceDao.getList(sql.toString(), new Object[]{upperTableName});
				for (int i = 0; i < tempDte.getRows(); i++) {
					tempFse = tempDte.getFieldSetEntity(i);
					singleFieldInfoObj = new JSONObject();
					fieldInfoObj.put(tempFse.getString("column_name"), singleFieldInfoObj);
					singleFieldInfoObj.put(CmnConst.TYPE, tempFse.getString("data_type").contains("VARCHAR") ? "varchar" : tempFse.getString("data_type"));
					singleFieldInfoObj.put(CmnConst.INTEGER, StringUtils.isEmpty(tempFse.getString("data_precision"))
							? tempFse.getString("char_length") : tempFse.getString("data_precision"));
					singleFieldInfoObj.put(CmnConst.DECIMAL, tempFse.getString("data_scale"));
					singleFieldInfoObj.put(CmnConst.NULLABLE, "N".equalsIgnoreCase(tempFse.getString("nullable")) ? 0 : 1);
					singleFieldInfoObj.put(CmnConst.COMMENT, tempFse.getString("comments"));
				}
				// 索引表
				sql.setLength(0);
				sql.append("SELECT DIC.INDEX_NAME,WM_CONCAT(DIC.COLUMN_NAME) column_name FROM SYS.DBA_IND_COLUMNS DIC");
				sql.append("\nLEFT JOIN SYS.DBA_INDEXES DI ON DIC.INDEX_NAME=DI.INDEX_NAME");
				sql.append("\nWHERE UNIQUENESS='NONUNIQUE' AND DIC.TABLE_OWNER=? AND DI.TABLE_NAME=?");
				sql.append("\nGROUP BY DIC.INDEX_NAME");
				tempDte = sourceDao.getList(sql.toString(), new Object[]{sourceDbName, upperTableName});
				for (int i = 0; i < tempDte.getRows(); i++) {
					tempFse = tempDte.getFieldSetEntity(i);
					singleFieldInfoObj = new JSONObject();
					indexInfoObj.put(tempFse.getString("index_name"), singleFieldInfoObj);
					singleFieldInfoObj.put(CmnConst.TYPE, CmnConst.NORMAL);
					singleFieldInfoObj.put(CmnConst.COLUMN_NAME, tempFse.getString("column_name"));
				}
				// 约束表 C-检查，写到字段里面；R-外键，不要；P-主键；U-唯一键
				sql.setLength(0);
				sql.append("SELECT DC.CONSTRAINT_NAME index_name,CONSTRAINT_TYPE,LISTAGG(COLUMN_NAME, ',') WITHIN GROUP(ORDER BY DCC.POSITION) column_name FROM SYS.DBA_CONS_COLUMNS DCC");
				sql.append("\nLEFT JOIN SYS.DBA_CONSTRAINTS DC ON DCC.CONSTRAINT_NAME=DC.CONSTRAINT_NAME");
				sql.append("\nWHERE DCC.OWNER=? AND DCC.TABLE_NAME=? AND CONSTRAINT_TYPE IN ('P','U')");
				sql.append("\nGROUP BY DC.CONSTRAINT_NAME,CONSTRAINT_TYPE");
				tempDte = sourceDao.getList(sql.toString(), new Object[]{sourceDbName, upperTableName});
				WriteUtil.append("DA-DDL-：" + sql + " |||库名：" + sourceDbName + " |||表名：" + upperTableName);
				for (int i = 0; i < tempDte.getRows(); i++) {
					tempFse = tempDte.getFieldSetEntity(i);
					singleFieldInfoObj = new JSONObject();
					indexInfoObj.put(tempFse.getString("index_name"), singleFieldInfoObj);
					singleFieldInfoObj.put(CmnConst.TYPE, "P".equalsIgnoreCase(tempFse.getString("constraint_type")) ? CmnConst.PRIMARY : CmnConst.UNIQUE);
					singleFieldInfoObj.put(CmnConst.COLUMN_NAME, tempFse.getString("column_name"));
				}
				WriteUtil.append("DA-DDL-创表信息：" + tableInfoObj);
			} else {
				throw new BaseException(ErrorCode.CREATE_TARGET_TABLE_FAIL);
			}
			return tableInfoObj;
		}

		/**
		 * 根据表结构信息，拼接DDL创建表sql语句
		 *
		 * @param tableInfoObj
		 * @return
		 */
		private String getSql(JSONObject tableInfoObj) {
			StringBuilder sql = new StringBuilder(128);
			JSONObject fieldInfoObj = tableInfoObj.getJSONObject(CmnConst.FIELD);
			JSONObject indexInfoObj = tableInfoObj.getJSONObject(CmnConst.INDEX);
			JSONObject singleFieldInfoObj;
			sql.append("CREATE TABLE ").append(tableInfoObj.getString(CmnConst.NAME)).append(" (");
			if (DataBaseType.MYSQL.equals(targetDao.getDataBaseType())) {
				// mysql
				for (String field : fieldInfoObj.keySet()) {
					singleFieldInfoObj = fieldInfoObj.getJSONObject(field);
					sql.append("\n    `").append(field.toLowerCase(Locale.ROOT)).append("` ");
					if (singleFieldInfoObj.getString(CmnConst.TYPE).contains("timestamp") || singleFieldInfoObj.getString(CmnConst.TYPE).contains("TIMESTAMP")) {
						sql.append("timestamp ");
					} else if (singleFieldInfoObj.getIntValue(CmnConst.DECIMAL) > 0) {
						sql.append("decimal(").append(singleFieldInfoObj.getString(CmnConst.INTEGER)).append(",").append(singleFieldInfoObj.getString(CmnConst.DECIMAL)).append(") ");
					} else if (singleFieldInfoObj.getString(CmnConst.TYPE).contains("int")) {
						sql.append("int(0) ");
					} else if ("number".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						if ("0".equals(singleFieldInfoObj.getString(CmnConst.DECIMAL))) {
							sql.append("int(").append(singleFieldInfoObj.getIntValue(CmnConst.INTEGER)).append(") ");
						} else {
							sql.append("decimal(22,4) ");
						}
					} else if ("date".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						sql.append("datetime(0) ");
					} else if ("blob".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						sql.append("blob ");
					} else if ("text".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE)) || singleFieldInfoObj.getString(CmnConst.TYPE).toLowerCase(Locale.ROOT).endsWith("text") ||
							(("varchar".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE)) || "varchar2".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) && singleFieldInfoObj.getIntValue(CmnConst.INTEGER) >= 4000)
					) {
						sql.append("text ");
					} else {
						sql.append(singleFieldInfoObj.getString(CmnConst.TYPE).toLowerCase(Locale.ROOT)).append("(").append(singleFieldInfoObj.getIntValue(CmnConst.INTEGER)).append(") ");
					}
					if ("varchar".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE)) || "varchar2".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE)) || "text".equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						sql.append(" CHARACTER SET utf8mb4 COLLATE utf8mb4_bin ");
					}
					if ("0".equals(singleFieldInfoObj.getString(CmnConst.NULLABLE))) {
						sql.append("not null ");
					}
					if (!StringUtils.isEmpty(singleFieldInfoObj.getString(CmnConst.COMMENT))) {
						sql.append(" comment '").append(singleFieldInfoObj.getString(CmnConst.COMMENT)).append("'");
					}
					sql.append(",");
				}
				for (String indexName : indexInfoObj.keySet()) {
					singleFieldInfoObj = indexInfoObj.getJSONObject(indexName);
					if (CmnConst.PRIMARY.equalsIgnoreCase(indexName) || CmnConst.PRIMARY.equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						sql.append("\n    PRIMARY KEY (`").append(singleFieldInfoObj.getString(CmnConst.COLUMN_NAME)).append("`) USING BTREE,");
					} else if (CmnConst.UNIQUE.equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						sql.append("\n    UNIQUE INDEX `").append(indexName).append("`(").append(singleFieldInfoObj.getString(CmnConst.COLUMN_NAME)).append(") USING BTREE,");
					} else if (CmnConst.NORMAL.equalsIgnoreCase(singleFieldInfoObj.getString(CmnConst.TYPE))) {
						sql.append("\n    INDEX `").append(indexName).append("`(").append(singleFieldInfoObj.getString(CmnConst.COLUMN_NAME)).append(") USING BTREE,");
					}
				}
				sql.deleteCharAt(sql.length() - 1);
				sql.append("\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ");
				if (!StringUtils.isEmpty(tableInfoObj.getString(CmnConst.COMMENT))) {
					sql.append(" COMMENT = '").append(tableInfoObj.getString(CmnConst.COMMENT)).append("'");
				}
			} else {
				throw new BaseException(ErrorCode.CREATE_TARGET_TABLE_FAIL);
			}
			return sql.toString();
		}

		private String saveCreateTableRecord(String tableName, String time) {
			/*=====================================================*/
			//                 新增创建表记录
			FieldSetEntity fse = new FieldSetEntity();
			fse.setTableName(CmnConst.DATA_ARCHIVING_SUB_TABLE);
			fse.setValue("table_name", tableName);
			fse.setValue("parent_uuid", this.configUid);
			fse.setValue("data_time", time);
			BaseUtil.createCreatorAndCreationTime(fse);
			getBaseDao().saveFieldSetEntity(fse);
			/*=====================================================*/
			return tableName;
		}


		private void createIndex(String targetTable, String time) throws BaseException {
			StringBuilder sql = new StringBuilder();
			sql.append("\n SELECT DBMS_METADATA.GET_DDL('INDEX',u.index_name) as  create_index_statement,u.INDEX_NAME ");
			sql.append("\n from USER_INDEXES u where u.TABLE_NAME=? ");
			String primaryIndexName = getPrimaryIndexName();
			List<Object> params = new ArrayList<>();
			params.add(sourceTable);
			if (!StringUtils.isEmpty(primaryIndexName)) {
				sql.append(" and u.index_name  != ? ");
				params.add(primaryIndexName);
			}
			DataTableEntity list = sourceDao.getList(sql.toString(), params.toArray());
			if (!DataTableEntity.isEmpty(list)) {
				for (int i = 0; i < list.getRows(); i++) {
					//循环获取创建索引语句
					String createIndexStatement = list.getString(i, "create_index_statement");
					String indexName = list.getString(i, "index_name");
					if (!StringUtils.isEmpty(createIndexStatement)) {
						//新的索引名称
						String newIndexName = this.getIndexName(indexName, time);
						//将建索引ddl
						createIndexStatement = createIndexStatement.replaceAll(sourceTable, targetTable)
								.replaceAll("\"" + indexName + "\"", "\"" + newIndexName + "\"");
						//执行创建索引
						targetDao.executeSql(createIndexStatement);
					}
				}
			}

		}

		/**
		 * 获取新的索引名称
		 *
		 * @param indexName
		 * @return
		 */
		private String getIndexName(String indexName, String time) {
			//新的索引名称
			String newIndexName;
			if (indexName.length() <= 26) {
				newIndexName = indexName + time;
			} else {
				//超过26位随机生成索引名称
				newIndexName = RandomUtil.randomString(10) + time;
			}
			return newIndexName;
		}

		/**
		 * 获取主键索引的名名称
		 *
		 * @return
		 */
		private String getPrimaryIndexName() {
			StringBuilder sql = new StringBuilder();
			sql.append("\n SELECT a.index_name from user_constraints a ");
			sql.append("\n WHERE a.constraint_type = 'P' ");
			sql.append("\n AND a.table_name = ? ");
			FieldSetEntity one = sourceDao.getOne(sql.toString(), new Object[]{this.sourceTable.toUpperCase()});
			return one != null ? one.getString("index_name") : null;
		}

		/**
		 * 获取创建表语句(包含主键索引)
		 *
		 * @return
		 */
		private String getCreateTableStatement() {
			StringBuilder sql = new StringBuilder();
			sql.append(" SELECT DBMS_METADATA.GET_DDL(U.OBJECT_TYPE, u.object_name) create_table_statement ");
			sql.append(" from USER_OBJECTS u ");
			sql.append(" where U.OBJECT_TYPE ='TABLE' and u.object_name=? ");
			FieldSetEntity one = sourceDao.getOne(sql.toString(), new Object[]{this.sourceTable.toUpperCase()});
			return one != null ? one.getString("create_table_statement") : null;
		}


		/**
		 * 获取其他索引名称
		 *
		 * @return
		 */
		private List<String> getOtherIndexName() {
			StringBuilder sql = new StringBuilder();
			sql.append(" SELECT INDEX_NAME FROM USER_INDEXES ");
			sql.append("\n WHERE TABLE_NAME=? AND  INDEX_NAME NOT IN ( ");
			sql.append("\n SELECT a.index_name from user_constraints a ");
			sql.append("\n WHERE a.constraint_type = 'P' ");
			sql.append("\n AND a.TABLE_NAME = ? ) ");

			DataTableEntity list = sourceDao.getList(sql.toString(), new Object[]{this.sourceTable, this.sourceTable});
			if (!DataTableEntity.isEmpty(list)) {
				return list.getData().stream().map(item -> item.getString("index_name")).collect(Collectors.toList());
			}
			return null;
		}


	}

}