public class CangoException extends RuntimeException {
    public CangoException() {
        super();
    }

    public CangoException(String message) {
        super(message);
    }

    public CangoException(String message, Throwable cause) {
        super(message, cause);
    }

    public CangoException(Throwable cause) {
        super(cause);
    }

    protected CangoException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
        super(message, cause, enableSuppression, writableStackTrace);
    }
} public class ColumnMeta {

    private String name;
    private int type;

    public ColumnMeta(String columnName, int columnType) {
        this.name = StringUtils.upperCase(columnName);// 统一为大写
        this.type = columnType;
    }

    public String getName() {
        return name;
    }

    public int getType() {
        return type;
    }

    public void setName(String name) {
        this.name = StringUtils.upperCase(name);
    }

    public void setType(int type) {
        this.type = type;
    }

    public ColumnMeta clone() {
        return new ColumnMeta(this.name, this.type);
    }

    public String toString() {
        return ToStringBuilder.reflectionToString(this, YuGongToStringStyle.DEFAULT_STYLE);
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((name == null) ? 0 : name.hashCode());
        result = prime * result + type;
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj == null) return false;
        if (getClass() != obj.getClass()) return false;
        ColumnMeta other = (ColumnMeta) obj;
        if (name == null) {
            if (other.name != null) return false;
        } else if (!name.equals(other.name)) return false;
        if (type != other.type) return false;
        return true;
    }

} public class ColumnValue {

    private ColumnMeta column;
    private Object value;
    private boolean check = true; // 是否需要做数据对比

    public ColumnValue() {
    }

    public ColumnValue(ColumnMeta column, Object value) {
        this(column, value, true);
    }

    public ColumnValue(ColumnMeta column, Object value, boolean check) {
        this.value = value;
        this.column = column;
        this.check = check;
    }

    public ColumnMeta getColumn() {
        return column;
    }

    public void setColumn(ColumnMeta column) {
        this.column = column;
    }

    public Object getValue() {
        return value;
    }

    public void setValue(Object value) {
        this.value = value;
    }

    public boolean isCheck() {
        return check;
    }

    public void setCheck(boolean check) {
        this.check = check;
    }

    public ColumnValue clone() {
        ColumnValue column = new ColumnValue();
        column.setValue(this.value);
        column.setColumn(this.column.clone());
        return column;
    }

    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((column == null) ? 0 : column.hashCode());
        result = prime * result + ((value == null) ? 0 : value.hashCode());
        return result;
    }

    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj == null) return false;
        if (getClass() != obj.getClass()) return false;
        ColumnValue other = (ColumnValue) obj;
        if (column == null) {
            if (other.column != null) return false;
        } else if (!column.equals(other.column)) return false;
        if (value == null) {
            if (other.value != null) return false;
        } else if (!value.equals(other.value)) return false;
        return true;
    }

    @Override
    public String toString() {
        return "ColumnValue [column=" + column + ", value=" + value + "]";
    }

} public class Table {

    private String type;
    private String schema;
    private String name;
    private List<ColumnMeta> primaryKeys = Lists.newArrayList();
    private List<ColumnMeta> columns = Lists.newArrayList();
    // 增量必带的扩展字段,比如DRDS模式下的拆分键
    private String extKey;

    public Table(String type, String schema, String name) {
        this.type = type;
        this.schema = schema;
        this.name = name;
    }

    public Table(String type, String schema, String name, List<ColumnMeta> primaryKeys, List<ColumnMeta> columns) {
        this.type = type;
        this.schema = schema;
        this.name = name;
        this.primaryKeys = primaryKeys;
        this.columns = columns;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public List<ColumnMeta> getPrimaryKeys() {
        return primaryKeys;
    }

    public void setPrimaryKeys(List<ColumnMeta> primaryKeys) {
        this.primaryKeys = primaryKeys;
    }

    public void addPrimaryKey(ColumnMeta primaryKey) {
        this.primaryKeys.add(primaryKey);
    }

    public List<ColumnMeta> getColumns() {
        return columns;
    }

    public void setColumns(List<ColumnMeta> columns) {
        this.columns = columns;
    }

    public void addColumn(ColumnMeta column) {
        this.columns.add(column);
    }

    /**
     * 返回所有字段信息，包括主键
     */
    public List<ColumnMeta> getColumnsWithPrimary() {
        List<ColumnMeta> result = Lists.newArrayList(primaryKeys);
        result.addAll(columns);
        return result;
    }

    public boolean isPrimaryKey(String columnName) {
        for (ColumnMeta col : primaryKeys) {
            if (StringUtils.equalsIgnoreCase(col.getName(), columnName)) {
                return true;
            }
        }

        return false;
    }

    /**
     * 返回schema.name
     */
    public String getFullName() {
        return schema + "." + name;
    }

    public String getExtKey() {
        return extKey;
    }

    public void setExtKey(String extKey) {
        this.extKey = extKey;
    }

    public String toString() {
        return ToStringBuilder.reflectionToString(this, YuGongToStringStyle.DEFAULT_STYLE);
    }

    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((name == null) ? 0 : name.hashCode());
        result = prime * result + ((schema == null) ? 0 : schema.hashCode());
        result = prime * result + ((type == null) ? 0 : type.hashCode());
        return result;
    }

    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj == null) return false;
        if (getClass() != obj.getClass()) return false;
        Table other = (Table) obj;
        if (name == null) {
            if (other.name != null) return false;
        } else if (!name.equals(other.name)) return false;
        if (schema == null) {
            if (other.schema != null) return false;
        } else if (!schema.equals(other.schema)) return false;
        if (type == null) {
            if (other.type != null) return false;
        } else if (!type.equals(other.type)) return false;
        return true;
    }

} public class TableMetaGenerator {

    private static final String mlogQuerySql = "select master,log_table from all_mview_logs where master = ?";
    private static final String mlogSchemaQuerySql = "select master,log_table from all_mview_logs where master = ? and log_owner = ?";
    private static final String queryShardKey = "show partitions from ?";

    /**
     * 获取对应的table meta信息，精确匹配
     *
     * @param dataSource
     * @param schemaName
     * @param tableName
     * @return
     */
    public static Table getTableMeta(final DataSource dataSource, final String schemaName, final String tableName) {
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        return (Table) jdbcTemplate.execute(new ConnectionCallback() {

            public Object doInConnection(Connection conn) throws SQLException, DataAccessException {
                DatabaseMetaData metaData = conn.getMetaData();
                String sName = getIdentifierName(schemaName, metaData);
                String tName = getIdentifierName(tableName, metaData);

                ResultSet rs = null;
                Table table;
                try {
                    rs = metaData.getTables(sName, sName, tName, new String[]{"TABLE"});
                    table = null;
                    while (rs.next()) {
                        String catlog = rs.getString(1);
                        String schema = rs.getString(2);
                        String name = rs.getString(3);
                        String type = rs.getString(4);

                        if ((sName == null || LikeUtil.isMatch(sName, catlog) || LikeUtil.isMatch(sName, schema))
                                && LikeUtil.isMatch(tName, name)) {
                            table = new Table(type, StringUtils.isEmpty(catlog) ? schema : catlog, name);
                            break;
                        }
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }

                if (table == null) {
                    throw new YuGongException("table[" + schemaName + "." + tableName + "] is not found");
                }

                // 查询所有字段
                List<ColumnMeta> columnList;
                try {
                    rs = metaData.getColumns(sName, sName, tName, null);
                    columnList = new ArrayList<ColumnMeta>();
                    while (rs.next()) {
                        String catlog = rs.getString(1);
                        String schema = rs.getString(2);
                        String name = rs.getString(3);
                        if ((sName == null || LikeUtil.isMatch(sName, catlog) || LikeUtil.isMatch(sName, schema))
                                && LikeUtil.isMatch(tName, name)) {
                            String columnName = rs.getString(4); // COLUMN_NAME
                            int columnType = rs.getInt(5);
                            String typeName = rs.getString(6);
                            columnType = convertSqlType(columnType, typeName);
                            ColumnMeta col = new ColumnMeta(columnName, columnType);
                            columnList.add(col);
                        }
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }

                // 查询主键信息
                List<String> primaryKeys = new ArrayList<String>();
                try {
                    rs = metaData.getPrimaryKeys(sName, sName, tName);
                    while (rs.next()) {
                        String catlog = rs.getString(1);
                        String schema = rs.getString(2);
                        String name = rs.getString(3);
                        if ((sName == null || LikeUtil.isMatch(sName, catlog) || LikeUtil.isMatch(sName, schema))
                                && LikeUtil.isMatch(tName, name)) {
                            primaryKeys.add(StringUtils.upperCase(rs.getString(4)));
                        }
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }

                List<String> uniqueKeys = new ArrayList<String>();
                if (primaryKeys.isEmpty()) {
                    String lastIndexName = null;
                    try {
                        rs = metaData.getIndexInfo(sName, sName, tName, true, true);
                        while (rs.next()) {
                            String catlog = rs.getString(1);
                            String schema = rs.getString(2);
                            String name = rs.getString(3);
                            if ((sName == null || LikeUtil.isMatch(sName, catlog) || LikeUtil.isMatch(sName, schema))
                                    && LikeUtil.isMatch(tName, name)) {
                                String indexName = StringUtils.upperCase(rs.getString(6));
                                if ("PRIMARY".equals(indexName)) {
                                    continue;
                                }

                                if (lastIndexName == null) {
                                    lastIndexName = indexName;
                                } else if (!lastIndexName.equals(indexName)) {
                                    break;
                                }

                                uniqueKeys.add(StringUtils.upperCase(rs.getString(9)));
                            }
                        }
                    } finally {
                        JdbcUtils.closeResultSet(rs);
                    }

                    // 如果无主键，使用唯一键
                    primaryKeys.addAll(uniqueKeys);
                }

                Set<ColumnMeta> columns = new HashSet<ColumnMeta>();
                Set<ColumnMeta> pks = new HashSet<ColumnMeta>();
                for (ColumnMeta columnMeta : columnList) {
                    if (primaryKeys.contains(columnMeta.getName())) {
                        pks.add(columnMeta);
                    } else {
                        columns.add(columnMeta);
                    }
                }

                table.getColumns().addAll(columns);
                table.getPrimaryKeys().addAll(pks);
                return table;
            }

        });
    }

    /**
     * 查询所有的表，不返回表中的字段
     *
     * @param dataSource
     * @return
     */
    public static List<Table> getTableMetasWithoutColumn(final DataSource dataSource, final String schemaName,
                                                         final String tableName) {
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        return (List<Table>) jdbcTemplate.execute(new ConnectionCallback() {

            public Object doInConnection(Connection conn) throws SQLException, DataAccessException {
                DatabaseMetaData metaData = conn.getMetaData();
                List<Table> result = Lists.newArrayList();
                String databaseName = metaData.getDatabaseProductName();
                String sName = getIdentifierName(schemaName, metaData);
                String tName = getIdentifierName(tableName, metaData);
                ResultSet rs = null;
                Table table = null;
                if (StringUtils.startsWithIgnoreCase(databaseName, "oracle") && StringUtils.isEmpty(schemaName)
                        && StringUtils.isEmpty(tableName)) {
                    // 针对oracle，只查询用户表，忽略系统表
                    Statement stmt = null;
                    try {
                        stmt = conn.createStatement();
                        rs = stmt.executeQuery("SELECT SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') AS SCHEMA_NAME , TABLE_NAME FROM USER_TABLES T , USER_USERS U WHERE U.USERNAME = SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA')");

                        while (rs.next()) {
                            String schema = rs.getString(1);
                            String name = rs.getString(2);
                            if (!StringUtils.startsWithIgnoreCase(name, "MLOG$_")) {
                                table = new Table("TABLE", schema, name);
                                result.add(table);
                            }
                        }
                    } finally {
                        JdbcUtils.closeResultSet(rs);
                        JdbcUtils.closeStatement(stmt);
                    }

                    return result;
                } else {
                    try {
                        rs = metaData.getTables(sName, sName, tName, new String[]{"TABLE"});
                        while (rs.next()) {
                            String catlog = rs.getString(1);
                            String schema = rs.getString(2);
                            String name = rs.getString(3);
                            String type = rs.getString(4);

                            if (!StringUtils.startsWithIgnoreCase(name, "MLOG$_")) {
                                table = new Table(type, StringUtils.isEmpty(catlog) ? schema : catlog, name);
                                result.add(table);
                            }
                        }
                    } finally {
                        JdbcUtils.closeResultSet(rs);
                    }
                    return result;
                }
            }

        });
    }

    /**
     * 返回就诶过 key:column name , value=index name
     */
    public static Map<String/* column name */, String /* index name */> getTableIndex(final DataSource dataSource,
                                                                                      final String schemaName,
                                                                                      final String tableName) {
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        return (Map<String, String>) jdbcTemplate.execute(new ConnectionCallback() {

            public Object doInConnection(Connection conn) throws SQLException, DataAccessException {
                DatabaseMetaData metaData = conn.getMetaData();
                String sName = getIdentifierName(schemaName, metaData);
                String tName = getIdentifierName(tableName, metaData);

                ResultSet rs = null;
                Map<String, String> indexes;
                try {
                    rs = metaData.getIndexInfo(sName, sName, tName, false, true);
                    indexes = new HashMap<String, String>();
                    while (rs.next()) {
                        String columnName = rs.getString(9);
                        String indexName = rs.getString(6);
                        if (columnName != null && indexName != null) {
                            indexes.put(columnName, indexName);
                        }
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }
                return indexes;
            }
        });
    }

    /**
     * <pre>
     * 常见的物化视图创建语句：
     * 1. CREATE MATERIALIZED VIEW LOG ON test_all_target with primary key;
     *
     * 本方法，主要提取生成物化视图的表名
     * </pre>
     *
     * @param dataSource
     * @param schemaName
     * @param tableName
     * @return
     */
    public static String getMLogTableName(final DataSource dataSource, final String schemaName, final String tableName) {
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        String querySql = StringUtils.isEmpty(schemaName) ? mlogQuerySql : mlogSchemaQuerySql;

        return (String) jdbcTemplate.execute(querySql, new PreparedStatementCallback() {

            public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
                DatabaseMetaData metaData = ps.getConnection().getMetaData();
                String sName = getIdentifierName(schemaName, metaData);
                String tName = getIdentifierName(tableName, metaData);

                ps.setString(1, tName);
                if (StringUtils.isNotEmpty(schemaName)) {
                    ps.setString(2, sName);
                }

                ResultSet rs = null;
                String log;
                try {
                    rs = ps.executeQuery();
                    log = null;
                    if (rs.next()) {
                        log = rs.getString("log_table");
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }
                return log;
            }
        });
    }

    public static void buildColumns(DataSource dataSource, final Table table) {
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        jdbcTemplate.execute(new ConnectionCallback() {

            public Object doInConnection(Connection conn) throws SQLException, DataAccessException {
                DatabaseMetaData metaData = conn.getMetaData();
                ResultSet rs = null;
                // 查询所有字段
                List<ColumnMeta> columnList;
                try {
                    rs = metaData.getColumns(table.getSchema(), table.getSchema(), table.getName(), null);
                    columnList = new ArrayList<ColumnMeta>();

                    while (rs.next()) {
                        String catlog = rs.getString(1);
                        String schema = rs.getString(2);
                        String name = rs.getString(3);
                        if ((table.getSchema() == null || LikeUtil.isMatch(table.getSchema(), catlog) || LikeUtil.isMatch(table.getSchema(),
                                schema))
                                && LikeUtil.isMatch(table.getName(), name)) {
                            String columnName = rs.getString(4); // COLUMN_NAME
                            int columnType = rs.getInt(5);
                            String typeName = rs.getString(6);
                            columnType = convertSqlType(columnType, typeName);
                            ColumnMeta col = new ColumnMeta(columnName, columnType);
                            columnList.add(col);
                        }
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }

                // 查询主键信息
                List<String> primaryKeys;
                try {
                    rs = metaData.getPrimaryKeys(table.getSchema(), table.getSchema(), table.getName());
                    primaryKeys = new ArrayList<String>();
                    while (rs.next()) {
                        String catlog = rs.getString(1);
                        String schema = rs.getString(2);
                        String name = rs.getString(3);
                        if ((table.getSchema() == null || StringUtils.equalsIgnoreCase(catlog, table.getSchema()) || StringUtils.equalsIgnoreCase(schema,
                                table.getSchema()))
                                && StringUtils.equalsIgnoreCase(name, table.getName())) {
                            primaryKeys.add(rs.getString(4));
                        }
                    }
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }

                Set<ColumnMeta> columns = new HashSet<ColumnMeta>();
                Set<ColumnMeta> pks = new HashSet<ColumnMeta>();
                for (ColumnMeta columnMeta : columnList) {
                    if (primaryKeys.contains(columnMeta.getName())) {
                        pks.add(columnMeta);
                    } else {
                        columns.add(columnMeta);
                    }
                }

                table.getColumns().addAll(columns);
                table.getPrimaryKeys().addAll(pks);
                return null;
            }

        });

    }

    /**
     * 获取DRDS下表的拆分字段, 返回格式为 id,name
     *
     * @param dataSource
     * @param schemaName
     * @param tableName
     * @return
     */
    public static String getShardKeyByDRDS(final DataSource dataSource, final String schemaName, final String tableName) {
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        return (String) jdbcTemplate.execute(queryShardKey, new PreparedStatementCallback() {

            public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
                DatabaseMetaData metaData = ps.getConnection().getMetaData();
                // String sName = getIdentifierName(schemaName, metaData);
                String tName = getIdentifierName(tableName, metaData);

                ps.setString(1, tName);
                ResultSet rs = null;
                try {
                    rs = ps.executeQuery();
                    String log = null;
                    if (rs.next()) {
                        log = rs.getString("KEYS");
                    }

                    return log;
                } finally {
                    JdbcUtils.closeResultSet(rs);
                }
            }
        });
    }

    /**
     * 根据{@linkplain java.sql.DatabaseMetaData}获取正确的表名
     * <p/>
     * <pre>
     * metaData中的storesUpperCaseIdentifiers，storesUpperCaseQuotedIdentifiers，storesLowerCaseIdentifiers,
     * storesLowerCaseQuotedIdentifiers,storesMixedCaseIdentifiers,storesMixedCaseQuotedIdentifiers
     * </pre>
     *
     * @param name
     * @param metaData
     * @return
     * @throws java.sql.SQLException
     */
    private static String getIdentifierName(String name, DatabaseMetaData metaData) throws SQLException {
        if (metaData.storesMixedCaseIdentifiers()) {
            return name; // 保留原始名
        } else if (metaData.storesUpperCaseIdentifiers()) {
            return StringUtils.upperCase(name);
        } else if (metaData.storesLowerCaseIdentifiers()) {
            return StringUtils.lowerCase(name);
        } else {
            return name;
        }
    }

    private static int convertSqlType(int columnType, String typeName) {
        String[] typeSplit = typeName.split(" ");
        if (typeSplit.length > 1) {
            if (columnType == Types.INTEGER && StringUtils.equalsIgnoreCase(typeSplit[1], "UNSIGNED")) {
                columnType = Types.BIGINT;
            }
        }

        if (columnType == Types.OTHER) {
            if (StringUtils.equalsIgnoreCase(typeName, "NVARCHAR")
                    || StringUtils.equalsIgnoreCase(typeName, "NVARCHAR2")) {
                columnType = Types.VARCHAR;
            }

            if (StringUtils.equalsIgnoreCase(typeName, "NCLOB")) {
                columnType = Types.CLOB;
            }

            if (StringUtils.startsWithIgnoreCase(typeName, "TIMESTAMP")) {
                columnType = Types.TIMESTAMP;
            }
        }
        return columnType;
    }

} public class SqlTemplates {

    public static SqlTemplate COMMON = new SqlTemplate();
} public enum ExtractStatus {
    /**
     * 正常提取
     */
    NORMAL,
    /**
     * 追上
     */
    CATCH_UP,
    /**
     * 无变更
     */
    NO_UPDATE,
    /**
     * 处理结束
     */
    TABLE_END;
} public class YuGongContext {

    // 具体每张表的同步
    private boolean ignoreSchema = false;  // 同步时是否忽略schema，oracle迁移到mysql可能schema不同，可设置为忽略

    // 全局共享
    private RunMode runMode;
    private int onceCrawNum;                   // 每次提取的记录数
    private int tpsLimit = 0;      // <=0代表不限制
    private DataSource sourceDs;                      // 源数据库链接
    private boolean batchApply = false;
    private boolean skipApplierException = false;  // 是否允许跳过applier异常
    private String sourceEncoding = "UTF-8";
    private DbType sourceDbType = DbType.ORACLE;

    private String dbName;
    private String dbHost;
    private Integer dbPort;
    private String name;

    public int getOnceCrawNum() {
        return onceCrawNum;
    }

    public void setOnceCrawNum(int onceCrawNum) {
        this.onceCrawNum = onceCrawNum;
    }

    public DataSource getSourceDs() {
        return sourceDs;
    }

    public void setSourceDs(DataSource sourceDs) {
        this.sourceDs = sourceDs;
    }


    public boolean isBatchApply() {
        return batchApply;
    }

    public void setBatchApply(boolean batchApply) {
        this.batchApply = batchApply;
    }

    public String getSourceEncoding() {
        return sourceEncoding;
    }

    public void setSourceEncoding(String sourceEncoding) {
        this.sourceEncoding = sourceEncoding;
    }

    public int getTpsLimit() {
        return tpsLimit;
    }

    public void setTpsLimit(int tpsLimit) {
        this.tpsLimit = tpsLimit;
    }

    public RunMode getRunMode() {
        return runMode;
    }

    public void setRunMode(RunMode runMode) {
        this.runMode = runMode;
    }

    public boolean isIgnoreSchema() {
        return ignoreSchema;
    }

    public void setIgnoreSchema(boolean ignoreSchema) {
        this.ignoreSchema = ignoreSchema;
    }

    public boolean isSkipApplierException() {
        return skipApplierException;
    }

    public void setSkipApplierException(boolean skipApplierException) {
        this.skipApplierException = skipApplierException;
    }

    public String getDbHost() {
        return dbHost;
    }

    public void setDbHost(String dbHost) {
        this.dbHost = dbHost;
    }

    public Integer getDbPort() {
        return dbPort;
    }

    public void setDbPort(Integer dbPort) {
        this.dbPort = dbPort;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getDbName() {
        return dbName;
    }

    public void setDbName(String dbName) {
        this.dbName = dbName;
    }

    public DbType getSourceDbType() {
        return sourceDbType;
    }

    public void setSourceDbType(DbType sourceDbType) {
        this.sourceDbType = sourceDbType;
    }

} public enum IncrementOpType {
    I/* INSERT */, U/* UPDATE */, D/* DELETE */;
} public class KafkaIncrementRecord extends IncrementRecord {
    private ColumnValue rowId;
    private Long executeTime;
    private List<ColumnValue> beforeColumns = Lists.newArrayList(); // 变动之前的值
    private List<ColumnValue> afterColumns = Lists.newArrayList(); // 变动之后的值

    public KafkaIncrementRecord() {
    }

    public KafkaIncrementRecord(String schemaName, String tableName, List<ColumnValue> primaryKeys, List<ColumnValue> columns) {
        super(schemaName, tableName, primaryKeys, columns);
    }

    public ColumnValue getRowId() {
        return rowId;
    }

    public void setRowId(ColumnValue rowId) {
        this.rowId = rowId;
    }

    public Long getExecuteTime() {
        return executeTime;
    }

    public void setExecuteTime(Long executeTime) {
        this.executeTime = executeTime;
    }

    public List<ColumnValue> getBeforeColumns() {
        return beforeColumns;
    }

    public void setBeforeColumns(List<ColumnValue> beforeColumns) {
        this.beforeColumns = beforeColumns;
    }

    public List<ColumnValue> getAfterColumns() {
        return afterColumns;
    }

    public void setAfterColumns(List<ColumnValue> afterColumns) {
        this.afterColumns = afterColumns;
    }

    @Override
    public KafkaIncrementRecord clone() {
        KafkaIncrementRecord record = new KafkaIncrementRecord();
        super.clone();
        record.setRowId(this.rowId);
        record.setExecuteTime(this.executeTime);
        record.setBeforeColumns(new ArrayList<ColumnValue>(this.beforeColumns));
        record.setAfterColumns(new ArrayList<ColumnValue>(this.afterColumns));
        return record;
    }
} public class YuGongException extends NestableRuntimeException {

    private static final long serialVersionUID = -654893533794556357L;

    public YuGongException(String errorCode) {
        super(errorCode);
    }

    public YuGongException(String errorCode, Throwable cause) {
        super(errorCode, cause);
    }

    public YuGongException(String errorCode, String errorDesc) {
        super(errorCode + ":" + errorDesc);
    }

    public YuGongException(String errorCode, String errorDesc, Throwable cause) {
        super(errorCode + ":" + errorDesc, cause);
    }

    public YuGongException(Throwable cause) {
        super(cause);
    }


}
import com.bella.cango.exception.CangoException;
import com.bella.cango.instance.oracle.common.db.meta.ColumnMeta;
import com.bella.cango.instance.oracle.common.db.meta.ColumnValue;
import com.bella.cango.instance.oracle.common.db.meta.Table;
import com.bella.cango.instance.oracle.common.db.meta.TableMetaGenerator;
import com.bella.cango.instance.oracle.common.db.sql.SqlTemplates;
import com.bella.cango.instance.oracle.common.model.ExtractStatus;
import com.bella.cango.instance.oracle.common.model.YuGongContext;
import com.bella.cango.instance.oracle.common.model.record.IncrementOpType;
import com.bella.cango.instance.oracle.common.model.record.KafkaIncrementRecord;
import com.bella.cango.instance.oracle.exception.YuGongException;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCallback;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.util.CollectionUtils;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.text.MessageFormat;
import java.util.*;
package com.bella.cango.instance.oracle.extractor;



public class KafkaIncRecordExtractor extends AbstractOracleRecordExtractor {
    private static final String MLOG_EXTRACT_FORMAT = "select rowid,{0} from {1}.{2} where rownum <= ? order by SEQUENCE$$ asc";
    private static final String MLOG_CLEAN_FORMAT = "delete from {0}.{1} where rowid=?";
    private static final String SINGLE_EXTRACT_FORMAT = "select rowid,{0} from {1}.{2} where SEQUENCE$$ = ?";
    private static final String CREATE_MLOG_FORMAT = "CREATE MATERIALIZED VIEW LOG ON {0}.{1} WITH SEQUENCE({2}), {3} INCLUDING NEW VALUES ";
    private YuGongContext context;
    private ColumnMeta rowidColumn = new ColumnMeta("rowid", Types.ROWID);
    private long sleepTime = 1000L;

    public KafkaIncRecordExtractor(YuGongContext context) {
        this.context = context;
    }

    public void start() {
        super.start();
    }

    public void stop() {
        super.stop();
    }

    @Override
    public Map<String, Object> extract(Table table) throws YuGongException {